Commit 699a371b authored by Dominik Charousset's avatar Dominik Charousset

zero timeouts

parent 6027a851
...@@ -73,7 +73,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -73,7 +73,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
return normal_exit_signal; return normal_exit_signal;
} }
} }
else if (v0 == atom(":Timeout")) { else if (v0 == atom("TIMEOUT")) {
return (v1 == m_active_timeout_id) ? timeout_message return (v1 == m_active_timeout_id) ? timeout_message
: expired_timeout_message; : expired_timeout_message;
} }
...@@ -87,10 +87,16 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -87,10 +87,16 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
void request_timeout(const util::duration& d) { void request_timeout(const util::duration& d) {
if (d.valid()) { if (d.valid()) {
get_scheduler()->future_send(this, d, atom(":Timeout"), ++m_active_timeout_id); if (d.is_zero()) {
// immediately enqueue timeout
enqueue(nullptr, make_any_tuple(atom("TIMEOUT"), ++m_active_timeout_id));
}
else {
get_scheduler()->future_send(this, d, atom("TIMEOUT"), ++m_active_timeout_id);
m_has_pending_timeout_request = true; m_has_pending_timeout_request = true;
} }
} }
}
void reset_timeout() { void reset_timeout() {
if (m_has_pending_timeout_request) { if (m_has_pending_timeout_request) {
......
...@@ -74,7 +74,7 @@ class duration { ...@@ -74,7 +74,7 @@ class duration {
constexpr duration() : unit(time_unit::none), count(0) { constexpr duration() : unit(time_unit::none), count(0) {
} }
constexpr duration(time_unit un, std::uint32_t val) : unit(un), count(val) { constexpr duration(time_unit tu, std::uint32_t val) : unit(tu), count(val) {
} }
template<class Rep, class Period> template<class Rep, class Period>
...@@ -84,11 +84,21 @@ class duration { ...@@ -84,11 +84,21 @@ class duration {
"only seconds, milliseconds or microseconds allowed"); "only seconds, milliseconds or microseconds allowed");
} }
template<class Rep>
constexpr duration(std::chrono::duration<Rep, std::ratio<60,1> > d)
: unit(time_unit::seconds), count(d.count() * 60) {
}
/** /**
* @brief Returns true if <tt>unit != time_unit::none</tt>. * @brief Returns true if <tt>unit != time_unit::none</tt>.
*/ */
inline bool valid() const { return unit != time_unit::none; } inline bool valid() const { return unit != time_unit::none; }
/**
* @brief Returns true if <tt>count == 0</tt>.
*/
inline bool is_zero() const { return count == 0; }
time_unit unit; time_unit unit;
std::uint32_t count; std::uint32_t count;
......
...@@ -58,13 +58,9 @@ void converted_thread_context::cleanup(std::uint32_t reason) { ...@@ -58,13 +58,9 @@ void converted_thread_context::cleanup(std::uint32_t reason) {
} }
void converted_thread_context::enqueue(actor* sender, any_tuple msg) { void converted_thread_context::enqueue(actor* sender, any_tuple msg) {
# ifdef CPPA_DEBUG
auto node = fetch_node(sender, std::move(msg)); auto node = fetch_node(sender, std::move(msg));
CPPA_REQUIRE(node->marked == false); CPPA_REQUIRE(node->marked == false);
m_mailbox.push_back(node); m_mailbox.push_back(node);
# else
m_mailbox.push_back(fetch_node(sender, std::move(msg)));
# endif
} }
void converted_thread_context::dequeue(partial_function& fun) { // override void converted_thread_context::dequeue(partial_function& fun) { // override
...@@ -84,19 +80,26 @@ void converted_thread_context::dequeue(behavior& bhvr) { // override ...@@ -84,19 +80,26 @@ void converted_thread_context::dequeue(behavior& bhvr) { // override
converted_thread_context::dequeue(fun); converted_thread_context::dequeue(fun);
} }
else if (invoke_from_cache(fun) == false) { else if (invoke_from_cache(fun) == false) {
if (bhvr.timeout().is_zero()) {
for (auto e = m_mailbox.try_pop(); e != nullptr; e = m_mailbox.try_pop()) {
CPPA_REQUIRE(e->marked == false);
if (invoke(e, bhvr)) return;
e = m_mailbox.try_pop();
}
bhvr.handle_timeout();
}
else {
auto timeout = now(); auto timeout = now();
timeout += bhvr.timeout(); timeout += bhvr.timeout();
recursive_queue_node* e = m_mailbox.try_pop(timeout); recursive_queue_node* e = m_mailbox.try_pop(timeout);
while (e != nullptr) { while (e != nullptr) {
if (e->marked) {
std::cout << "ooops: " << to_string(e->msg) << std::endl;
}
CPPA_REQUIRE(e->marked == false); CPPA_REQUIRE(e->marked == false);
if (invoke(e, fun)) return; if (invoke(e, fun)) return;
e = m_mailbox.try_pop(timeout); e = m_mailbox.try_pop(timeout);
} }
bhvr.handle_timeout(); bhvr.handle_timeout();
} }
}
} }
filter_result converted_thread_context::filter_msg(const any_tuple& msg) { filter_result converted_thread_context::filter_msg(const any_tuple& msg) {
......
...@@ -99,16 +99,25 @@ void yielding_actor::dequeue(behavior& bhvr) { ...@@ -99,16 +99,25 @@ void yielding_actor::dequeue(behavior& bhvr) {
yielding_actor::dequeue(fun); yielding_actor::dequeue(fun);
} }
else if (invoke_from_cache(bhvr) == false) { else if (invoke_from_cache(bhvr) == false) {
if (bhvr.timeout().is_zero()) {
for (auto e = m_mailbox.try_pop(); e != nullptr; e = m_mailbox.try_pop()) {
CPPA_REQUIRE(e->marked == false);
if (invoke(e, bhvr)) return;
}
bhvr.handle_timeout();
}
else {
request_timeout(bhvr.timeout()); request_timeout(bhvr.timeout());
recursive_queue_node* e = nullptr; recursive_queue_node* e = nullptr;
do { do {
e = m_mailbox.try_pop(); e = m_mailbox.try_pop();
while (!e) { while (!e) {
CPPA_REQUIRE(e->marked == false);
yield_until_not_empty(); yield_until_not_empty();
e = m_mailbox.try_pop(); e = m_mailbox.try_pop();
} }
} while (invoke(e, bhvr) == false);
} }
while (invoke(e, bhvr) == false);
} }
} }
......
#define CPPA_VERBOSE_CHECK
#include <stack> #include <stack>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
...@@ -261,7 +263,8 @@ std::string behavior_test(actor_ptr et) { ...@@ -261,7 +263,8 @@ std::string behavior_test(actor_ptr et) {
on_arg_match >> [&](const std::string& str) { on_arg_match >> [&](const std::string& str) {
result = str; result = str;
}, },
after(std::chrono::seconds(2)) >> [&]() { after(std::chrono::minutes(1)) >> [&]() {
//after(std::chrono::seconds(2)) >> [&]() {
throw std::runtime_error(testee_name + " does not reply"); throw std::runtime_error(testee_name + " does not reply");
} }
); );
...@@ -372,6 +375,18 @@ size_t test__spawn() { ...@@ -372,6 +375,18 @@ size_t test__spawn() {
receive(on(1, 2, 3) >> []() { }); receive(on(1, 2, 3) >> []() { });
CPPA_IF_VERBOSE(cout << "ok" << endl); CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "test receive with zero timeout ... " << std::flush);
receive (
others() >> []() {
cerr << "WTF?? received: " << to_string(self->last_dequeued())
<< endl;
},
after(std::chrono::seconds(0)) >> []() {
// mailbox empty
}
);
CPPA_IF_VERBOSE(cout << "ok" << endl);
auto mirror = actor_prototype ( auto mirror = actor_prototype (
others() >> []() { others() >> []() {
self->last_sender() << self->last_dequeued(); self->last_sender() << self->last_dequeued();
......
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