Commit 963a0bc8 authored by Dominik Charousset's avatar Dominik Charousset

Use `unique_ptr` when dealing with `message_data`

parent 96a251b3
...@@ -78,6 +78,10 @@ class message_data : public ref_counted { ...@@ -78,6 +78,10 @@ class message_data : public ref_counted {
ptr& operator=(ptr&&) = default; ptr& operator=(ptr&&) = default;
ptr& operator=(const ptr&) = default; ptr& operator=(const ptr&) = default;
inline ptr(const std::nullptr_t&) {
// nop
}
inline explicit ptr(message_data* p) : m_ptr(p) {} inline explicit ptr(message_data* p) : m_ptr(p) {}
inline void detach() { static_cast<void>(get_detached()); } inline void detach() { static_cast<void>(get_detached()); }
......
...@@ -80,7 +80,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -80,7 +80,7 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from priority policy // member functions from priority policy
unique_mailbox_element_pointer next_message() { mailbox_element_ptr next_message() {
return priority_policy().next_message(dptr()); return priority_policy().next_message(dptr());
} }
...@@ -88,7 +88,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -88,7 +88,7 @@ class proper_actor_base : public Policies::resume_policy::template
return priority_policy().has_next_message(dptr()); return priority_policy().has_next_message(dptr());
} }
void push_to_cache(unique_mailbox_element_pointer ptr) { void push_to_cache(mailbox_element_ptr ptr) {
priority_policy().push_to_cache(dptr(), std::move(ptr)); priority_policy().push_to_cache(dptr(), std::move(ptr));
} }
......
...@@ -528,7 +528,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -528,7 +528,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
} }
template <class... Ts> template <class... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) { inline mailbox_element_ptr new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...); return mailbox_element::create(std::forward<Ts>(args)...);
} }
......
...@@ -30,11 +30,8 @@ ...@@ -30,11 +30,8 @@
#include "caf/mixin/memory_cached.hpp" #include "caf/mixin/memory_cached.hpp"
// needs access to constructor + destructor to initialize m_dummy_node
namespace caf { namespace caf {
class local_actor;
class mailbox_element : public extend<memory_managed>:: class mailbox_element : public extend<memory_managed>::
with<mixin::memory_cached> { with<mixin::memory_cached> {
public: public:
...@@ -46,6 +43,7 @@ class mailbox_element : public extend<memory_managed>:: ...@@ -46,6 +43,7 @@ class mailbox_element : public extend<memory_managed>::
message msg; // 'content field' message msg; // 'content field'
mailbox_element(); mailbox_element();
mailbox_element(actor_addr sender, message_id id);
mailbox_element(actor_addr sender, message_id id, message data); mailbox_element(actor_addr sender, message_id id, message data);
~mailbox_element(); ~mailbox_element();
...@@ -55,19 +53,16 @@ class mailbox_element : public extend<memory_managed>:: ...@@ -55,19 +53,16 @@ class mailbox_element : public extend<memory_managed>::
mailbox_element& operator=(mailbox_element&&) = delete; mailbox_element& operator=(mailbox_element&&) = delete;
mailbox_element& operator=(const mailbox_element&) = delete; mailbox_element& operator=(const mailbox_element&) = delete;
template <class T> using unique_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
static mailbox_element* create(actor_addr sender, message_id id, T&& data) {
return detail::memory::create<mailbox_element>(std::move(sender), id, static unique_ptr create(actor_addr sender, message_id id, message msg);
std::forward<T>(data));
}
inline bool is_high_priority() const { inline bool is_high_priority() const {
return mid.is_high_priority(); return mid.is_high_priority();
} }
}; };
using unique_mailbox_element_pointer = using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
std::unique_ptr<mailbox_element, detail::disposer>;
} // namespace caf } // namespace caf
......
...@@ -298,7 +298,7 @@ class message { ...@@ -298,7 +298,7 @@ class message {
m_vals.detach(); m_vals.detach();
} }
void reset(); void reset(data_ptr new_ptr = nullptr);
explicit message(raw_ptr); explicit message(raw_ptr);
......
...@@ -34,13 +34,11 @@ namespace caf { ...@@ -34,13 +34,11 @@ namespace caf {
namespace policy { namespace policy {
class cooperative_scheduling { class cooperative_scheduling {
public: public:
using timeout_type = int; using timeout_type = int;
template <class Actor> template <class Actor>
inline void launch(Actor* self, execution_unit* host, bool lazy) { void launch(Actor* self, execution_unit* host, bool lazy) {
// detached in scheduler::worker::run // detached in scheduler::worker::run
self->attach_to_scheduler(); self->attach_to_scheduler();
if (lazy) { if (lazy) {
...@@ -57,15 +55,15 @@ class cooperative_scheduling { ...@@ -57,15 +55,15 @@ class cooperative_scheduling {
template <class Actor> template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid, void enqueue(Actor* self, const actor_addr& sender, message_id mid,
message& msg, execution_unit* eu) { message& msg, execution_unit* eu) {
auto e = self->new_mailbox_element(sender, mid, std::move(msg)); auto ptr = self->new_mailbox_element(sender, mid, std::move(msg));
switch (self->mailbox().enqueue(e)) { switch (self->mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: { case detail::enqueue_result::unblocked_reader: {
// re-schedule actor // re-schedule actor
if (eu) if (eu) {
eu->exec_later(self); eu->exec_later(self);
else } else {
detail::singletons::get_scheduling_coordinator()->enqueue( detail::singletons::get_scheduling_coordinator()->enqueue(self);
self); }
break; break;
} }
case detail::enqueue_result::queue_closed: { case detail::enqueue_result::queue_closed: {
...@@ -80,7 +78,6 @@ class cooperative_scheduling { ...@@ -80,7 +78,6 @@ class cooperative_scheduling {
break; break;
} }
} }
}; };
} // namespace policy } // namespace policy
......
...@@ -56,7 +56,7 @@ class no_scheduling { ...@@ -56,7 +56,7 @@ class no_scheduling {
message& msg, execution_unit*) { message& msg, execution_unit*) {
auto ptr = self->new_mailbox_element(sender, mid, std::move(msg)); auto ptr = self->new_mailbox_element(sender, mid, std::move(msg));
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr)) { if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr.release())) {
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer srb{self->exit_reason()}; detail::sync_request_bouncer srb{self->exit_reason()};
srb(sender, mid); srb(sender, mid);
......
...@@ -34,8 +34,8 @@ namespace policy { ...@@ -34,8 +34,8 @@ namespace policy {
class not_prioritizing { class not_prioritizing {
public: public:
template <class Actor> template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self) { mailbox_element_ptr next_message(Actor* self) {
return unique_mailbox_element_pointer{self->mailbox().try_pop()}; return mailbox_element_ptr{self->mailbox().try_pop()};
} }
template <class Actor> template <class Actor>
...@@ -44,7 +44,7 @@ class not_prioritizing { ...@@ -44,7 +44,7 @@ class not_prioritizing {
} }
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) { void push_to_cache(Actor* self, mailbox_element_ptr ptr) {
self->mailbox().cache().push_second_back(ptr.release()); self->mailbox().cache().push_second_back(ptr.release());
} }
......
...@@ -78,7 +78,7 @@ class prioritizing { ...@@ -78,7 +78,7 @@ class prioritizing {
} }
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) { void push_to_cache(Actor* self, mailbox_element_ptr ptr) {
auto high_prio = [](const mailbox_element& val) { auto high_prio = [](const mailbox_element& val) {
return val.is_high_priority(); return val.is_high_priority();
}; };
......
...@@ -35,7 +35,7 @@ class priority_policy { ...@@ -35,7 +35,7 @@ class priority_policy {
* Returns the next message from the mailbox or `nullptr` if it's empty. * Returns the next message from the mailbox or `nullptr` if it's empty.
*/ */
template <class Actor> template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self); mailbox_element_ptr next_message(Actor* self);
/** /**
* Queries whether the mailbox is not empty. * Queries whether the mailbox is not empty.
...@@ -47,7 +47,7 @@ class priority_policy { ...@@ -47,7 +47,7 @@ class priority_policy {
* Stores the message in a cache for later retrieval. * Stores the message in a cache for later retrieval.
*/ */
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr); void push_to_cache(Actor* self, mailbox_element_ptr ptr);
/** /**
* Removes the first element from the cache matching predicate `p`. * Removes the first element from the cache matching predicate `p`.
......
...@@ -80,23 +80,23 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -80,23 +80,23 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
timer_actor_policies>, timer_actor_policies>,
public spawn_as_is { public spawn_as_is {
public: public:
inline unique_mailbox_element_pointer dequeue() { inline mailbox_element_ptr dequeue() {
await_data(); await_data();
return next_message(); return next_message();
} }
inline unique_mailbox_element_pointer try_dequeue(const hrc::time_point& tp) { inline mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (scheduling_policy().await_data(this, tp)) { if (scheduling_policy().await_data(this, tp)) {
return next_message(); return next_message();
} }
return unique_mailbox_element_pointer{}; return mailbox_element_ptr{};
} }
void act() override { void act() override {
trap_exit(true); trap_exit(true);
// setup & local variables // setup & local variables
bool done = false; bool done = false;
unique_mailbox_element_pointer msg_ptr; mailbox_element_ptr msg_ptr;
std::multimap<hrc::time_point, delayed_msg> messages; std::multimap<hrc::time_point, delayed_msg> messages;
// message handling rules // message handling rules
message_handler mfun{ message_handler mfun{
......
...@@ -28,6 +28,15 @@ mailbox_element::mailbox_element() ...@@ -28,6 +28,15 @@ mailbox_element::mailbox_element()
// nop // nop
} }
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(arg0)),
mid(arg1) {
// nop
}
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2) mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2)
: next(nullptr), : next(nullptr),
prev(nullptr), prev(nullptr),
...@@ -42,4 +51,11 @@ mailbox_element::~mailbox_element() { ...@@ -42,4 +51,11 @@ mailbox_element::~mailbox_element() {
// nop // nop
} }
mailbox_element_ptr mailbox_element::create(actor_addr sender, message_id id,
message msg) {
auto ptr = detail::memory::create<mailbox_element>(std::move(sender), id,
std::move(msg));
return mailbox_element_ptr{ptr};
}
} // namespace caf } // namespace caf
...@@ -46,8 +46,8 @@ message& message::operator=(message&& other) { ...@@ -46,8 +46,8 @@ message& message::operator=(message&& other) {
return *this; return *this;
} }
void message::reset() { void message::reset(data_ptr new_ptr) {
m_vals.reset(); m_vals.swap(new_ptr);
} }
void* message::mutable_at(size_t p) { void* message::mutable_at(size_t p) {
......
...@@ -361,7 +361,7 @@ class broker : public extend<local_actor>:: ...@@ -361,7 +361,7 @@ class broker : public extend<local_actor>::
policy::sequential_invoke m_invoke_policy; policy::sequential_invoke m_invoke_policy;
middleman& m_mm; middleman& m_mm;
std::deque<unique_mailbox_element_pointer> m_cache; std::deque<mailbox_element_ptr> m_cache;
}; };
class broker::functor_based : public extend<broker>:: class broker::functor_based : public extend<broker>::
......
...@@ -187,9 +187,9 @@ void broker::invoke_message(const actor_addr& sender, message_id mid, ...@@ -187,9 +187,9 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
break; break;
case policy::im_skipped: { case policy::im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg"); CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
auto e = mailbox_element::create(sender, bid, auto ptr = mailbox_element::create(sender, bid,
std::move(m_dummy_node.msg)); std::move(m_dummy_node.msg));
m_cache.push_back(unique_mailbox_element_pointer{e}); m_cache.push_back(std::move(ptr));
break; break;
} }
} }
......
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