Unverified Commit 00f7bb3a authored by Noir's avatar Noir Committed by GitHub

Merge branch 'master' into topic/neverlord/meta-objects-diagnostic

parents 0b56818f 26b0a900
......@@ -17,6 +17,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Typed actors that use a `typed_actor_pointer` can now access the
`run_{delayed,scheduled}` member functions.
- Scheduled and delayed sends now return a disposable (#1362).
- Typed response handles received support for converting them to observable or
single objects.
- Typed actors that use the type-erased pointer-view type received access to the
new flow API functions (e.g., `make_observable`).
- Not initializing the meta objects table now prints a diagnosis message before
aborting the program. Previously, the application would usually just crash due
to a `nullptr`-access inside some CAF function.
......@@ -36,6 +40,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
operators that use the `ucast` operator internally.
- The `mcast` and `ucast` operators now stop calling `on_next` immediately when
disposed.
- Actors no longer terminate despite having open streams (#1377).
## [0.19.0-rc.1] - 2022-10-31
......
......@@ -370,6 +370,7 @@ caf_add_component(
thread_hook
tracing_data
type_id_list
typed_actor_view
typed_behavior
typed_message_view
typed_response_promise
......
......@@ -249,8 +249,9 @@ public:
async::consumer_resource<T> to_resource() {
return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand);
}
const observable& as_observable() const& noexcept {
return std::move(*this);
return *this;
}
observable&& as_observable() && noexcept {
......
......@@ -142,8 +142,9 @@ public:
void fwd_on_complete(buffer_emit_t) {
control_sub_ = nullptr;
err_ = make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream");
if (state_ == state::running)
err_ = make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream");
shutdown();
}
......
......@@ -107,13 +107,15 @@ public:
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::single<T>> as_single() && {
static_assert(std::is_same_v<response_type, message>);
static_assert(std::is_same_v<response_type, detail::type_list<T>>
|| std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_);
}
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>> as_observable() && {
static_assert(std::is_same_v<response_type, message>);
static_assert(std::is_same_v<response_type, detail::type_list<T>>
|| std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_).as_observable();
}
......
......@@ -709,7 +709,7 @@ public:
bool alive() const noexcept {
return !bhvr_stack_.empty() || !awaited_responses_.empty()
|| !multiplexed_responses_.empty() || !watched_disposables_.empty()
|| !stream_sources_.empty();
|| !stream_sources_.empty() || !stream_bridges_.empty();
}
/// Runs all pending actions.
......
......@@ -8,12 +8,24 @@
#include "caf/config.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/none.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream.hpp"
#include "caf/timespan.hpp"
#include "caf/typed_actor_view_base.hpp"
#include "caf/typed_stream.hpp"
namespace caf {
/// Utility function to force the type of `self` to depend on `T` and to raise a
/// compiler error if the user did not include 'caf/scheduled_actor/flow.hpp'.
/// The function itself does nothing and simply returns `self`.
template <class T>
auto typed_actor_view_flow_access(caf::scheduled_actor* self) {
using Self = flow::assert_scheduled_actor_hdr_t<T, caf::scheduled_actor*>;
return static_cast<Self>(self);
}
/// Decorates a pointer to a @ref scheduled_actor with a statically typed actor
/// interface.
template <class... Sigs>
......@@ -296,6 +308,101 @@ public:
return self_;
}
// -- flow API ---------------------------------------------------------------
/// @copydoc flow::coordinator::make_observable
template <class T = none_t>
auto make_observable() {
// Note: the template parameter T serves no purpose other than forcing the
// compiler to delay evaluation of this function body by having
// *something* to pass to `typed_actor_view_flow_access`.
auto self = typed_actor_view_flow_access<T>(self_);
return self->make_observable();
}
/// @copydoc scheduled_actor::to_stream
template <class Observable>
auto to_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable&& obs) {
auto self = typed_actor_view_flow_access<Observable>(self_);
return self->to_stream(std::move(name), max_delay, max_items_per_batch,
std::forward<Observable>(obs));
}
/// @copydoc scheduled_actor::to_stream
template <class Observable>
auto to_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable&& obs) {
return to_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch, std::forward<Observable>(obs));
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_stream_t to_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) {
return {self_, std::move(name), max_delay, max_items_per_batch};
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_stream_t to_stream(std::string name, timespan max_delay,
size_t max_items_per_batch) {
return to_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch);
}
/// @copydoc scheduled_actor::to_typed_stream
template <class Observable>
auto to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
auto self = typed_actor_view_flow_access<Observable>(self_);
return self->to_typed_stream(std::move(name), max_delay,
max_items_per_batch, std::move(obs));
}
/// @copydoc scheduled_actor::to_typed_stream
template <class Observable>
auto to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
return to_typed_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch, std::move(obs));
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_typed_stream_t
to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) {
return {self_, std::move(name), max_delay, max_items_per_batch};
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_typed_stream_t
to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch) {
return to_typed_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch);
}
/// @copydoc scheduled_actor::observe
template <class T>
auto
observe(typed_stream<T> what, size_t buf_capacity, size_t demand_threshold) {
auto self = typed_actor_view_flow_access<T>(self_);
return self->observe(std::move(what), buf_capacity, demand_threshold);
}
/// @copydoc scheduled_actor::observe_as
template <class T>
auto observe_as(stream what, size_t buf_capacity, size_t demand_threshold) {
auto self = typed_actor_view_flow_access<T>(self_);
return self->template observe_as<T>(std::move(what), buf_capacity,
demand_threshold);
}
/// @copydoc scheduled_actor::deregister_stream
void deregister_stream(uint64_t stream_id) {
self_->deregister_stream(stream_id);
}
private:
scheduled_actor* self_;
};
......
......@@ -572,21 +572,45 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::internal;
}
case type_id_v<stream_open_msg>: {
// Try to subscribe the sink to the observable.
auto& [str_id, ptr, sink_id] = content.get_as<stream_open_msg>(0);
if (!ptr) {
CAF_LOG_ERROR("received a stream_open_msg with a null sink");
return message_category::internal;
}
auto sink_hdl = actor_cast<actor>(ptr);
if (auto i = stream_sources_.find(str_id); i != stream_sources_.end()) {
// Create a forwarder that turns observed items into batches.
auto fwd = make_counted<batch_forwarder_impl>(this, sink_hdl, sink_id);
auto sub = i->second.obs->subscribe(flow::observer<async::batch>{fwd});
if (fwd->subscribed()) {
// Inform the sink that the stream is now open.
auto flow_id = new_u64_id();
stream_subs_.emplace(flow_id, std::move(fwd));
auto mipb = static_cast<uint32_t>(i->second.max_items_per_batch);
unsafe_send_as(this, sink_hdl,
stream_ack_msg{ctrl(), sink_id, flow_id, mipb});
if (sink_hdl.node() != node()) {
// Actors cancel any pending streams when they terminate. However,
// remote actors may terminate without sending us a proper goodbye.
// Hence, we add a function object to remote actors to make sure we
// get a cancel in all cases.
auto weak_self = weak_actor_ptr{ctrl()};
sink_hdl->attach_functor([weak_self, flow_id] {
if (auto sptr = weak_self.lock())
anon_send(actor_cast<actor>(sptr), stream_cancel_msg{flow_id});
});
}
} else {
CAF_LOG_ERROR("failed to subscribe a batch forwarder");
sub.dispose();
}
} else {
// Abort the flow immediately.
CAF_LOG_DEBUG("requested stream does not exist");
auto err = make_error(sec::invalid_stream);
unsafe_send_as(this, sink_hdl,
stream_abort_msg{sink_id, std::move(err)});
}
return message_category::internal;
}
......@@ -600,6 +624,7 @@ scheduled_actor::categorize(mailbox_element& x) {
case type_id_v<stream_cancel_msg>: {
auto [sub_id] = content.get_as<stream_cancel_msg>(0);
if (auto i = stream_subs_.find(sub_id); i != stream_subs_.end()) {
CAF_LOG_DEBUG("canceled stream " << sub_id);
i->second->cancel();
stream_subs_.erase(i);
}
......
......@@ -14,6 +14,8 @@ using namespace caf;
namespace {
using i32_worker = typed_actor<result<int32_t>(int32_t)>;
struct dummy_state {
static inline const char* name = "dummy";
......@@ -33,9 +35,11 @@ using dummy_actor = stateful_actor<dummy_state>;
struct fixture : test_coordinator_fixture<> {
actor dummy;
i32_worker typed_dummy;
fixture() {
dummy = sys.spawn<dummy_actor>();
typed_dummy = actor_cast<i32_worker>(sys.spawn<dummy_actor>());
sched.run();
}
};
......@@ -45,7 +49,7 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("response handles are convertible to observables and singles") {
GIVEN("a response handle that produces a valid result") {
GIVEN("a response handle with dynamic typing that produces a valid result") {
WHEN("calling as_single") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
......@@ -89,7 +93,51 @@ SCENARIO("response handles are convertible to observables and singles") {
}
}
}
GIVEN("a response handle that produces an error") {
GIVEN("a response handle with static typing that produces a valid result") {
WHEN("calling as_single") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{42})
.as_single<int32_t>()
.subscribe([&result](int32_t val) { result = val; },
[&result](const error& what) { result = what; });
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(42));
expect((int32_t), from(typed_dummy).to(aut).with(84));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{84});
}
}
WHEN("calling as_observable") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
size_t on_next_calls = 0;
bool completed = false;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{42})
.as_observable<int32_t>()
.do_on_error([&](const error& what) { result = what; })
.do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val;
++on_next_calls;
});
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(42));
expect((int32_t), from(typed_dummy).to(aut).with(84));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{84});
CHECK_EQ(on_next_calls, 1u);
CHECK(completed);
}
}
}
GIVEN("a response handle with dynamic typing that produces an error") {
WHEN("calling as_single") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
......@@ -133,6 +181,50 @@ SCENARIO("response handles are convertible to observables and singles") {
}
}
}
GIVEN("a response handle with static typing that produces an error") {
WHEN("calling as_single") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{13})
.as_single<int32_t>()
.subscribe([&result](int32_t val) { result = val; },
[&result](const error& what) { result = what; });
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(13));
expect((error), from(typed_dummy).to(aut));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{make_error(sec::invalid_argument)});
}
}
WHEN("calling as_observable") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
size_t on_next_calls = 0;
bool completed = false;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{13})
.as_observable<int32_t>()
.do_on_error([&](const error& what) { result = what; })
.do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val;
++on_next_calls;
});
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(13));
expect((error), from(typed_dummy).to(aut));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{make_error(sec::invalid_argument)});
CHECK_EQ(on_next_calls, 0u);
CHECK(!completed);
}
}
}
}
END_FIXTURE_SCOPE()
// 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.
#define CAF_SUITE typed_actor_view
#include "caf/typed_actor_view.hpp"
#include "core-test.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/typed_actor.hpp"
using namespace caf;
using namespace std::literals;
namespace {
using shared_int = std::shared_ptr<int>;
using shared_err = std::shared_ptr<error>;
using int_actor = typed_actor<result<void>(int)>;
using int_actor_ptr = int_actor::pointer_view;
struct int_actor_state {
using init_fn = std::function<void(int_actor_ptr)>;
int_actor_state(int_actor_ptr ptr, init_fn fn)
: self(ptr), init(std::move(fn)) {
// nop
}
int_actor::behavior_type make_behavior() {
init(self);
return {
[](int) {},
};
}
int_actor_ptr self;
init_fn init;
};
using int_actor_impl = int_actor::stateful_impl<int_actor_state>;
void stream_observer(event_based_actor* self, stream str, shared_int res,
shared_err err) {
// Access `self` through the view to check correct forwarding of `observe_as`.
auto view = typed_actor_view<result<void>(int)>{self};
view.observe_as<int>(str, 30, 10)
.do_on_error([err](const error& what) { *err = what; })
.for_each([res](int value) { *res += value; });
}
void typed_stream_observer(event_based_actor* self, typed_stream<int> str,
shared_int res, shared_err err) {
// Access `self` through the view to check correct forwarding of `observe`.
auto view = typed_actor_view<result<void>(int)>{self};
view.observe(str, 30, 10)
.do_on_error([err](const error& what) { *err = what; })
.for_each([res](int value) { *res += value; });
}
struct fixture : test_coordinator_fixture<> {
int_actor spawn_int_actor(int_actor_state::init_fn init) {
return sys.spawn<int_actor_impl>(std::move(init));
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("typed actors may use the flow API") {
GIVEN("a typed actor") {
WHEN("the actor calls make_observable") {
THEN("the function returns a flow that lives on the typed actor") {
auto res = std::make_shared<int>(0);
spawn_int_actor([=](int_actor_ptr self) {
self
->make_observable() //
.iota(1)
.take(10)
.for_each([res](int value) { *res += value; });
});
run();
CHECK_EQ(*res, 55);
}
}
WHEN("the actor creates a stream via compose") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_stream("foo", 10ms, 10));
self->spawn(stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK_EQ(*err, sec::none);
}
}
WHEN("the actor creates a stream via to_stream") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto obs = self
->make_observable() //
.iota(1)
.take(10)
.as_observable();
auto str = self->to_stream("foo", 10ms, 10, obs);
self->spawn(stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
WHEN("the actor creates a typed stream via compose") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_typed_stream("foo", 10ms, 10));
self->spawn(typed_stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
WHEN("the actor creates a typed stream via to_typed_stream") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto obs = self
->make_observable() //
.iota(1)
.take(10)
.as_observable();
auto str = self->to_typed_stream("foo", 10ms, 10, obs);
self->spawn(typed_stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
}
WHEN("the actor creates a stream and deregisters it immediately") {
THEN("other actors receive an error when observing the stream") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_stream("foo", 10ms, 10));
self->spawn(stream_observer, str, res, err);
self->deregister_stream(str.id());
});
run();
CHECK_EQ(*res, 0);
CHECK_EQ(*err, sec::invalid_stream);
}
}
}
END_FIXTURE_SCOPE()
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