Commit ab400a9f authored by Dominik Charousset's avatar Dominik Charousset

Make mailbox countable

parent 8b28990a
...@@ -47,11 +47,6 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -47,11 +47,6 @@ class proper_actor_base : public Policies::resume_policy::template
CAF_REQUIRE(this->is_registered() == false); CAF_REQUIRE(this->is_registered() == false);
} }
// grant access to the actor's mailbox
typename Base::mailbox_type& mailbox() {
return this->m_mailbox;
}
// member functions from scheduling policy // member functions from scheduling policy
using timeout_type = typename Policies::scheduling_policy::timeout_type; using timeout_type = typename Policies::scheduling_policy::timeout_type;
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <mutex> #include <mutex>
#include <atomic> #include <atomic>
#include <memory> #include <memory>
#include <limits>
#include <condition_variable> // std::cv_status #include <condition_variable> // std::cv_status
#include "caf/config.hpp" #include "caf/config.hpp"
...@@ -192,6 +193,17 @@ class single_reader_queue { ...@@ -192,6 +193,17 @@ class single_reader_queue {
clear(); clear();
} }
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
size_t res = 0;
fetch_new_data();
auto ptr = m_head;
while (ptr && res < max_count) {
ptr = ptr->next;
++res;
}
return res;
}
/************************************************************************** /**************************************************************************
* support for synchronized access * * support for synchronized access *
**************************************************************************/ **************************************************************************/
...@@ -252,7 +264,6 @@ class single_reader_queue { ...@@ -252,7 +264,6 @@ class single_reader_queue {
// atomically sets m_stack back and enqueues all elements to the cache // atomically sets m_stack back and enqueues all elements to the cache
bool fetch_new_data(pointer end_ptr) { bool fetch_new_data(pointer end_ptr) {
CAF_REQUIRE(m_head == nullptr);
CAF_REQUIRE(end_ptr == nullptr || end_ptr == stack_empty_dummy()); CAF_REQUIRE(end_ptr == nullptr || end_ptr == stack_empty_dummy());
pointer e = m_stack.load(); pointer e = m_stack.load();
// must not be called on a closed queue // must not be called on a closed queue
......
...@@ -524,12 +524,12 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -524,12 +524,12 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
return &m_dummy_node; return &m_dummy_node;
} }
protected:
template <class... Ts> template <class... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) { inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...); return mailbox_element::create(std::forward<Ts>(args)...);
} }
protected:
// identifies the ID of the last sent synchronous request // identifies the ID of the last sent synchronous request
message_id m_last_request_id; message_id m_last_request_id;
......
...@@ -32,10 +32,9 @@ namespace mixin { ...@@ -32,10 +32,9 @@ namespace mixin {
template <class Base, class Subtype> template <class Base, class Subtype>
class mailbox_based : public Base { class mailbox_based : public Base {
using del = detail::disposer;
public: public:
using del = detail::disposer;
using mailbox_type = detail::single_reader_queue<mailbox_element, del>;
~mailbox_based() { ~mailbox_based() {
if (!m_mailbox.closed()) { if (!m_mailbox.closed()) {
...@@ -44,29 +43,25 @@ class mailbox_based : public Base { ...@@ -44,29 +43,25 @@ class mailbox_based : public Base {
} }
} }
template <class... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
void cleanup(uint32_t reason) { void cleanup(uint32_t reason) {
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{reason};
m_mailbox.close(f); m_mailbox.close(f);
Base::cleanup(reason); Base::cleanup(reason);
} }
protected: mailbox_type& mailbox() {
return m_mailbox;
}
protected:
using combined_type = mailbox_based; using combined_type = mailbox_based;
using mailbox_type = detail::single_reader_queue<mailbox_element, del>;
template <class... Ts> template <class... Ts>
mailbox_based(Ts&&... args) mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) {
: Base(std::forward<Ts>(args)...) {} // nop
}
mailbox_type m_mailbox; mailbox_type m_mailbox;
}; };
} // namespace mixin } // namespace mixin
......
...@@ -883,6 +883,17 @@ class actor_size_getter : public event_based_actor { ...@@ -883,6 +883,17 @@ class actor_size_getter : public event_based_actor {
} }
}; };
void counting_actor(event_based_actor* self) {
for (int i = 0; i < 100; ++i) {
self->send(self, atom("dummy"));
}
CAF_CHECK_EQUAL(self->mailbox().count(), 100);
for (int i = 0; i < 100; ++i) {
self->send(self, atom("dummy"));
}
CAF_CHECK_EQUAL(self->mailbox().count(), 200);
}
} // namespace <anonymous> } // namespace <anonymous>
int main() { int main() {
...@@ -890,6 +901,9 @@ int main() { ...@@ -890,6 +901,9 @@ int main() {
spawn<actor_size_getter>(); spawn<actor_size_getter>();
await_all_actors_done(); await_all_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
spawn(counting_actor);
await_all_actors_done();
CAF_CHECKPOINT();
test_spawn(); test_spawn();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
await_all_actors_done(); await_all_actors_done();
......
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