Commit 909ad46a authored by Dominik Charousset's avatar Dominik Charousset

Add new disposable callbacks (actions) abstraction

parent abd89335
......@@ -97,6 +97,7 @@ caf_add_component(
src/deserializer.cpp
src/detail/abstract_worker.cpp
src/detail/abstract_worker_hub.cpp
src/detail/action.cpp
src/detail/append_percent_encoded.cpp
src/detail/base64.cpp
src/detail/behavior_impl.cpp
......@@ -135,6 +136,7 @@ caf_add_component(
src/detail/tick_emitter.cpp
src/detail/token_based_credit_controller.cpp
src/detail/type_id_list_builder.cpp
src/disposable.cpp
src/downstream_manager.cpp
src/downstream_manager_base.cpp
src/error.cpp
......@@ -239,6 +241,7 @@ caf_add_component(
decorator.sequencer
deep_to_string
detached_actors
detail.action
detail.base64
detail.bounds_checker
detail.config_consumer
......
// 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/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <atomic>
namespace caf::detail {
/// A functional interface similar to `std::function<void()>` with dispose
/// semantics.
class CAF_CORE_EXPORT action {
public:
enum class state {
disposed,
scheduled,
invoked,
};
class impl : public ref_counted, public disposable::impl {
public:
friend class action;
impl();
void dispose() override;
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
/// Tries setting the state from `invoked` back to `scheduled`.
/// @return the new state.
state reschedule();
/// Runs the action if the state is `scheduled`.
virtual void run() = 0;
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref();
}
protected:
std::atomic<state> state_;
};
using impl_ptr = intrusive_ptr<impl>;
action(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
action() noexcept = default;
action(action&&) noexcept = default;
action(const action&) noexcept = default;
action& operator=(action&&) noexcept = default;
action& operator=(const action&) noexcept = default;
/// Runs the action if it is still scheduled for execution, does nothing
/// otherwise.
void run();
/// Cancel the action if it has not been invoked yet.
void dispose() {
pimpl_->dispose();
}
[[nodiscard]] bool disposed() const {
return pimpl_->state_ == state::disposed;
}
[[nodiscard]] bool scheduled() const {
return pimpl_->state_ == state::scheduled;
}
[[nodiscard]] bool invoked() const {
return pimpl_->state_ == state::invoked;
}
/// Tries setting the state from `invoked` back to `scheduled`.
/// @return the new state.
state reschedule() {
return pimpl_->reschedule();
}
/// Returns a pointer to the implementation.
[[nodiscard]] impl* ptr() noexcept {
return pimpl_.get();
}
/// Returns a pointer to the implementation.
[[nodiscard]] const impl* ptr() const noexcept {
return pimpl_.get();
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] disposable as_disposable() && noexcept {
return disposable{std::move(pimpl_)};
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] disposable as_disposable() const& noexcept {
return disposable{pimpl_};
}
private:
impl_ptr pimpl_;
};
template <class F>
action make_action(F f) {
struct impl : action::impl {
F f_;
explicit impl(F fn) : f_(std::move(fn)) {
// nop
}
void run() override {
if (state_ == action::state::scheduled) {
f_();
auto expected = action::state::scheduled;
// No retry. If this action has been disposed while running, we stay in
// the state 'disposed'.
state_.compare_exchange_strong(expected, action::state::invoked);
}
}
};
return action{make_counted<impl>(std::move(f))};
}
} // namespace caf::detail
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::detail::action)
......@@ -7,6 +7,7 @@
#include <algorithm>
#include <cstddef>
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
......@@ -16,6 +17,7 @@
#include "caf/detail/stringification_inspector.hpp"
#include "caf/inspector_access.hpp"
#include "caf/serializer.hpp"
#include "caf/type_id.hpp"
namespace caf::detail::default_function {
......@@ -56,9 +58,14 @@ bool load(deserializer& source, void* ptr) {
template <class T>
void stringify(std::string& buf, const void* ptr) {
stringification_inspector f{buf};
auto unused = f.apply(*static_cast<const T*>(ptr));
static_cast<void>(unused);
if constexpr (is_allowed_unsafe_message_type_v<T>) {
auto tn = type_name_v<T>;
buf.insert(buf.end(), tn.begin(), tn.end());
} else {
stringification_inspector f{buf};
auto unused = f.apply(*static_cast<const T*>(ptr));
static_cast<void>(unused);
}
}
} // namespace caf::detail::default_function
......
// 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/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
/// Represents a disposable resource.
class CAF_CORE_EXPORT disposable {
public:
/// Internal implementation class of a `disposable`.
class impl {
public:
virtual ~impl();
virtual void dispose() = 0;
virtual bool disposed() const noexcept = 0;
disposable as_disposable() noexcept;
virtual void ref_disposable() const noexcept = 0;
virtual void deref_disposable() const noexcept = 0;
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref_disposable();
}
};
explicit disposable(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
disposable() noexcept = default;
disposable(disposable&&) noexcept = default;
disposable(const disposable&) noexcept = default;
disposable& operator=(disposable&&) noexcept = default;
disposable& operator=(const disposable&) noexcept = default;
/// Disposes the resource. Calling `dispose()` on a disposed resource is a
/// no-op.
void dispose() {
if (pimpl_) {
pimpl_->dispose();
pimpl_ = nullptr;
}
}
/// Returns whether the resource has been disposed.
[[nodiscard]] bool disposed() const noexcept {
return pimpl_ ? pimpl_->disposed() : true;
}
/// Returns whether this handle still points to a resource.
[[nodiscard]] bool valid() const noexcept {
return pimpl_ != nullptr;
}
/// Returns `valid()`;
explicit operator bool() const noexcept {
return valid();
}
/// Returns `!valid()`;
bool operator!() const noexcept {
return !valid();
}
/// Returns a pointer to the implementation.
[[nodiscard]] impl* ptr() noexcept {
return pimpl_.get();
}
/// Returns a pointer to the implementation.
[[nodiscard]] const impl* ptr() const noexcept {
return pimpl_.get();
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl> as_intrusive_ptr() const& noexcept {
return pimpl_;
}
/// Exchanges the content of this handle with `other`.
void swap(disposable& other) {
pimpl_.swap(other.pimpl_);
}
private:
intrusive_ptr<impl> pimpl_;
};
} // namespace caf
......@@ -347,6 +347,7 @@ class stream_distribution_tree;
class abstract_worker;
class abstract_worker_hub;
class action;
class disposer;
class dynamic_message_data;
class group_manager;
......
......@@ -385,6 +385,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_TYPE_ID(core_module, (caf::actor_addr))
CAF_ADD_TYPE_ID(core_module, (caf::byte_buffer))
CAF_ADD_TYPE_ID(core_module, (caf::config_value))
CAF_ADD_TYPE_ID(core_module, (caf::detail::action))
CAF_ADD_TYPE_ID(core_module, (caf::dictionary<caf::config_value>) )
CAF_ADD_TYPE_ID(core_module, (caf::down_msg))
CAF_ADD_TYPE_ID(core_module, (caf::downstream_msg))
......
// 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/detail/action.hpp"
#include "caf/logger.hpp"
namespace caf::detail {
action::impl::impl() : state_(action::state::scheduled) {
// nop
}
void action::impl::dispose() {
state_ = state::disposed;
}
bool action::impl::disposed() const noexcept {
return state_.load() == state::disposed;
}
void action::impl::ref_disposable() const noexcept {
ref();
}
void action::impl::deref_disposable() const noexcept {
deref();
}
action::state action::impl::reschedule() {
auto expected = state::invoked;
if (state_.compare_exchange_strong(expected, state::scheduled))
return state::scheduled;
else
return expected;
}
void action::run() {
CAF_LOG_TRACE("");
CAF_ASSERT(pimpl_ != nullptr);
pimpl_->run();
}
} // namespace caf::detail
// 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/disposable.hpp"
namespace caf {
disposable::impl::~impl() {
// nop
}
disposable disposable::impl::as_disposable() noexcept {
return disposable{intrusive_ptr<disposable::impl>{this}};
}
} // namespace caf
......@@ -10,6 +10,7 @@
#include "caf/actor_system.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/config_value.hpp"
#include "caf/detail/action.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/error.hpp"
#include "caf/group.hpp"
......
// 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 detail.action
#include "caf/detail/action.hpp"
#include "core-test.hpp"
using namespace caf;
SCENARIO("actions wrap function calls") {
GIVEN("an action wrapping a lambda") {
WHEN("running the action") {
THEN("it calls the lambda and transitions from scheduled to invoked") {
auto called = false;
auto uut = detail::make_action([&called] { called = true; });
CHECK(uut.scheduled());
uut.run();
CHECK(called);
CHECK(uut.invoked());
}
}
WHEN("disposing the action") {
THEN("it transitions to disposed and run no longer calls the lambda") {
auto called = false;
auto uut = detail::make_action([&called] { called = true; });
CHECK(uut.scheduled());
uut.dispose();
CHECK(uut.disposed());
uut.run();
CHECK(!called);
CHECK(uut.disposed());
}
}
WHEN("running the action multiple times") {
THEN("any call after the first becomes a no-op") {
auto n = 0;
auto uut = detail::make_action([&n] { ++n; });
uut.run();
uut.run();
uut.run();
CHECK(uut.invoked());
CHECK_EQ(n, 1);
}
}
WHEN("re-scheduling an action after running it") {
THEN("then the lambda gets invoked twice") {
auto n = 0;
auto uut = detail::make_action([&n] { ++n; });
uut.run();
uut.run();
CHECK_EQ(uut.reschedule(), detail::action::state::scheduled);
uut.run();
uut.run();
CHECK(uut.invoked());
CHECK_EQ(n, 2);
}
}
WHEN("converting an action to a disposable") {
THEN("the disposable and the action point to the same impl object") {
auto uut = detail::make_action([] {});
auto d1 = uut.as_disposable(); // const& overload
auto d2 = detail::action{uut}.as_disposable(); // && overload
CHECK_EQ(uut.ptr(), d1.ptr());
CHECK_EQ(uut.ptr(), d2.ptr());
}
}
}
}
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