Commit e09e8634 authored by neverlord's avatar neverlord

event_based_actor improvements and fixes

parent 4ff15bf7
...@@ -31,6 +31,8 @@ class abstract_event_based_actor : public scheduled_actor ...@@ -31,6 +31,8 @@ class abstract_event_based_actor : public scheduled_actor
void resume(util::fiber*, resume_callback* callback) /*override*/; void resume(util::fiber*, resume_callback* callback) /*override*/;
virtual void on_exit() = 0;
private: private:
void handle_message(std::unique_ptr<queue_node>& node, void handle_message(std::unique_ptr<queue_node>& node,
......
...@@ -65,6 +65,15 @@ class scheduled_actor : public abstract_actor<local_actor> ...@@ -65,6 +65,15 @@ class scheduled_actor : public abstract_actor<local_actor>
void request_timeout(const util::duration& d); void request_timeout(const util::duration& d);
void reset_timeout()
{
if (m_has_pending_timeout_request)
{
++m_active_timeout_id;
m_has_pending_timeout_request = false;
}
}
private: private:
bool m_has_pending_timeout_request; bool m_has_pending_timeout_request;
......
...@@ -13,7 +13,7 @@ class event_based_actor : public detail::abstract_event_based_actor ...@@ -13,7 +13,7 @@ class event_based_actor : public detail::abstract_event_based_actor
become(std::move(rules)); become(std::move(rules));
} }
inline void become_impl(timed_invoke_rules& rules) inline void become_impl(timed_invoke_rules&& rules)
{ {
become(std::move(rules)); become(std::move(rules));
} }
...@@ -27,11 +27,11 @@ class event_based_actor : public detail::abstract_event_based_actor ...@@ -27,11 +27,11 @@ class event_based_actor : public detail::abstract_event_based_actor
protected: protected:
void unbecome() /*override*/; void unbecome();
void become(invoke_rules&& behavior) /*override*/; void become(invoke_rules&& behavior);
void become(timed_invoke_rules&& behavior) /*override*/; void become(timed_invoke_rules&& behavior);
template<typename Head, typename... Tail> template<typename Head, typename... Tail>
void become(invoke_rules&& rules, Head&& head, Tail&&... tail) void become(invoke_rules&& rules, Head&& head, Tail&&... tail)
...@@ -45,6 +45,8 @@ class event_based_actor : public detail::abstract_event_based_actor ...@@ -45,6 +45,8 @@ class event_based_actor : public detail::abstract_event_based_actor
virtual void init() = 0; virtual void init() = 0;
virtual void on_exit();
template<typename Scheduler> template<typename Scheduler>
abstract_event_based_actor* abstract_event_based_actor*
attach_to_scheduler(void (*enqueue_fun)(Scheduler*, scheduled_actor*), attach_to_scheduler(void (*enqueue_fun)(Scheduler*, scheduled_actor*),
...@@ -55,21 +57,35 @@ class event_based_actor : public detail::abstract_event_based_actor ...@@ -55,21 +57,35 @@ class event_based_actor : public detail::abstract_event_based_actor
return this; return this;
} }
private: protected:
// provoke compiler errors for usage of receive() and related functions // provoke compiler errors for usage of receive() and related functions
template<typename... Args> template<typename... Args>
void receive(Args&&... rules); void receive(Args&&...)
{
static_assert(sizeof...(Args) < 0,
"You shall not use receive in an event-based actor. "
"Use become()/unbecome() instead.");
}
template<typename... Args> template<typename... Args>
void receive_loop(Args&&... rules); void receive_loop(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename Statement> template<typename... Args>
void receive_while(Statement&& stmt); void receive_while(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename... Args> template<typename... Args>
void do_receive(Args&&... args); void do_receive(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
}; };
......
...@@ -141,6 +141,35 @@ class either ...@@ -141,6 +141,35 @@ class either
return *this; return *this;
} }
either& operator=(either&& other)
{
if (m_is_left == other.m_is_left)
{
if (m_is_left)
{
left() = std::move(other.left());
}
else
{
right() = std::move(other.right());
}
}
else
{
destroy();
m_is_left = other.m_is_left;
if (other.m_is_left)
{
cr_left(std::move(other.left()));
}
else
{
cr_right(std::move(other.right()));
}
}
return *this;
}
inline bool is_left() const inline bool is_left() const
{ {
return m_is_left; return m_is_left;
...@@ -153,21 +182,25 @@ class either ...@@ -153,21 +182,25 @@ class either
Left& left() Left& left()
{ {
check_flag(true, "left");
return m_left; return m_left;
} }
const Left& left() const const Left& left() const
{ {
check_flag(true, "left");
return m_left; return m_left;
} }
Right& right() Right& right()
{ {
check_flag(false, "right");
return m_right; return m_right;
} }
const Right& right() const const Right& right() const
{ {
check_flag(false, "right");
return m_right; return m_right;
} }
......
...@@ -16,23 +16,18 @@ class singly_linked_list ...@@ -16,23 +16,18 @@ class singly_linked_list
typedef T element_type; typedef T element_type;
singly_linked_list() : m_head(0), m_tail(0) { } singly_linked_list() : m_head(nullptr), m_tail(nullptr) { }
~singly_linked_list() ~singly_linked_list()
{ {
while (m_head) clear();
{
T* next = m_head->next;
delete m_head;
m_head = next;
}
} }
inline bool empty() const { return m_head == 0; } inline bool empty() const { return m_head == nullptr; }
void push_back(element_type* what) void push_back(element_type* what)
{ {
what->next = 0; what->next = nullptr;
if (m_tail) if (m_tail)
{ {
m_tail->next = what; m_tail->next = what;
...@@ -48,10 +43,21 @@ class singly_linked_list ...@@ -48,10 +43,21 @@ class singly_linked_list
{ {
element_type* first = m_head; element_type* first = m_head;
element_type* last = m_tail; element_type* last = m_tail;
m_head = m_tail = 0; m_head = m_tail = nullptr;
return { first, last }; return { first, last };
} }
void clear()
{
while (m_head)
{
T* next = m_head->next;
delete m_head;
m_head = next;
}
m_head = m_tail = nullptr;
}
}; };
} } // namespace cppa::util } } // namespace cppa::util
......
...@@ -14,25 +14,34 @@ void abstract_event_based_actor::dequeue(timed_invoke_rules&) ...@@ -14,25 +14,34 @@ void abstract_event_based_actor::dequeue(timed_invoke_rules&)
} }
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node, void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node,
invoke_rules& behavior) invoke_rules& behavior)
{ {
// no need to handle result // no need to handle result
(void) dq(node, behavior, m_buffer); (void) dq(node, behavior, m_buffer);
} }
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node, void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node,
timed_invoke_rules& behavior) timed_invoke_rules& behavior)
{ {
// request timeout only if we're running short on messages
if (m_mailbox.empty() && has_pending_timeout() == false)
{
request_timeout(behavior.timeout());
}
switch (dq(node, behavior, m_buffer)) switch (dq(node, behavior, m_buffer))
{ {
case dq_timeout_occured: case dq_timeout_occured:
{ {
behavior.handle_timeout(); behavior.handle_timeout();
// fall through
}
case dq_done:
{
// callback might have called become()/unbecome()
// request next timeout if needed
if (!m_loop_stack.empty())
{
if (m_loop_stack.top().is_right())
{
request_timeout(m_loop_stack.top().right().timeout());
}
}
break;
} }
default: break; default: break;
} }
...@@ -52,10 +61,12 @@ void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& nod ...@@ -52,10 +61,12 @@ void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& nod
void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback) void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
{ {
set_self(this);
auto done_cb = [&]() auto done_cb = [&]()
{ {
m_state.store(scheduled_actor::done); m_state.store(scheduled_actor::done);
while (!m_loop_stack.empty()) m_loop_stack.pop(); while (!m_loop_stack.empty()) m_loop_stack.pop();
on_exit();
callback->exec_done(); callback->exec_done();
}; };
......
...@@ -2,6 +2,10 @@ ...@@ -2,6 +2,10 @@
namespace cppa { namespace cppa {
void event_based_actor::on_exit()
{
}
void event_based_actor::unbecome() void event_based_actor::unbecome()
{ {
if (!m_loop_stack.empty()) m_loop_stack.pop(); if (!m_loop_stack.empty()) m_loop_stack.pop();
...@@ -9,11 +13,13 @@ void event_based_actor::unbecome() ...@@ -9,11 +13,13 @@ void event_based_actor::unbecome()
void event_based_actor::become(invoke_rules&& behavior) void event_based_actor::become(invoke_rules&& behavior)
{ {
reset_timeout();
m_loop_stack.push(std::move(behavior)); m_loop_stack.push(std::move(behavior));
} }
void event_based_actor::become(timed_invoke_rules&& behavior) void event_based_actor::become(timed_invoke_rules&& behavior)
{ {
request_timeout(behavior.timeout());
m_loop_stack.push(std::move(behavior)); m_loop_stack.push(std::move(behavior));
} }
......
...@@ -133,7 +133,11 @@ scheduled_actor::dq_result scheduled_actor::dq(std::unique_ptr<queue_node>& node ...@@ -133,7 +133,11 @@ scheduled_actor::dq_result scheduled_actor::dq(std::unique_ptr<queue_node>& node
// m_active_timeout_id is already invalid // m_active_timeout_id is already invalid
m_has_pending_timeout_request = false; m_has_pending_timeout_request = false;
// restore mailbox before calling client // restore mailbox before calling client
if (!buffer.empty()) m_mailbox.push_front(std::move(buffer)); if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
return dq_timeout_occured; return dq_timeout_occured;
} }
default: break; default: break;
...@@ -144,7 +148,11 @@ scheduled_actor::dq_result scheduled_actor::dq(std::unique_ptr<queue_node>& node ...@@ -144,7 +148,11 @@ scheduled_actor::dq_result scheduled_actor::dq(std::unique_ptr<queue_node>& node
m_last_dequeued = std::move(node->msg); m_last_dequeued = std::move(node->msg);
m_last_sender = std::move(node->sender); m_last_sender = std::move(node->sender);
// restore mailbox before invoking imd // restore mailbox before invoking imd
if (!buffer.empty()) m_mailbox.push_front(std::move(buffer)); if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
// expire pending request // expire pending request
if (m_has_pending_timeout_request) if (m_has_pending_timeout_request)
{ {
......
...@@ -25,55 +25,78 @@ using namespace cppa; ...@@ -25,55 +25,78 @@ using namespace cppa;
class event_testee : public event_based_actor class event_testee : public event_based_actor
{ {
invoke_rules wait4string()
{
return on<std::string>() >> [=](const std::string& value)
{
cout << "event_testee[string]: " << value << endl;
// switch back to wait4int
unbecome();
unbecome();
};
}
invoke_rules wait4float()
{
return on<float>() >> [=](float value)
{
cout << "event_testee[float]: " << value << endl;
become(wait4string());
};
}
invoke_rules wait4int()
{
return on<int>() >> [=](int value)
{
cout << "event_testee[int]: " << value << endl;
become(wait4float());
};
}
public: public:
void on_exit()
{
cout << "event_testee::on_exit()" << endl;
}
void init() void init()
{ {
become cout << "event_testee::init()" << endl;
( become(wait4int());
on<int>() >> [&](int i)
{
// do NOT call receive() here!
// this would hijack the worker thread
become
(
on<int>() >> [&](int i2)
{
cout << "event testee: (" << i << ", " << i2 << ")" << endl;
unbecome();
},
on<float>() >> [&](float f)
{
cout << "event testee: (" << i << ", " << f << ")" << endl;
become
(
on<float>() >> [&]()
{
// switch back to the outer behavior
unbecome();
unbecome();
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(last_received())
<< endl;
}
);
}
);
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(last_received())
<< endl;
}
);
} }
}; };
event_based_actor* event_testee2()
{
struct impl : event_based_actor
{
int num_timeouts;
impl() : num_timeouts(0) { }
void init()
{
become
(
others() >> []()
{
cout << "event testee2: "
<< to_string(last_received())
<< endl;
},
after(std::chrono::milliseconds(50)) >> [this]()
{
cout << "testee2 received timeout nr. "
<< (num_timeouts + 1) << endl;
if (++num_timeouts >= 5) unbecome();
}
);
}
};
return new impl;
}
class testee_behavior : public actor_behavior class testee_behavior : public actor_behavior
{ {
...@@ -81,12 +104,27 @@ class testee_behavior : public actor_behavior ...@@ -81,12 +104,27 @@ class testee_behavior : public actor_behavior
void act() void act()
{ {
cout << "testee_behavior::act()" << endl;
receive_loop receive_loop
( (
after(std::chrono::milliseconds(10)) >> []() on<int>() >> [&](int i)
{ {
quit(exit_reason::user_defined); cout << "testee_behavior[int]: " << i << endl;
receive
(
on<float>() >> [&](float f)
{
cout << "testee_behavior[float]: " << f << endl;
receive
(
on<std::string>() >> [&](const std::string& str)
{
cout << "testee_behavior[string]: "
<< str
<< endl;
}
);
}
);
} }
); );
} }
...@@ -142,27 +180,41 @@ void testee3(actor_ptr parent) ...@@ -142,27 +180,41 @@ void testee3(actor_ptr parent)
); );
} }
template<class Testee>
void behavior_test()
{
std::string testee_name = detail::to_uniform_name(typeid(Testee));
cout << "behavior_test<" << testee_name << ">()" << endl;
auto et = spawn(new Testee);
send(et, 1);
send(et, 2);
send(et, 3);
send(et, .1f);
send(et, "hello " + testee_name);
send(et, .2f);
send(et, .3f);
send(et, "hello again " + testee_name);
send(et, "goodbye " + testee_name);
send(et, atom(":Exit"), exit_reason::user_defined);
await_all_others_done();
cout << endl;
}
size_t test__spawn() size_t test__spawn()
{ {
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
spawn(testee1); //spawn(testee1);
spawn(new testee_behavior); //spawn(new testee_behavior);
await_all_others_done(); //await_all_others_done();
auto et = spawn(new event_testee); //spawn(event_testee2());
send(et, 42);
send(et, 24);
send(et, 42);
send(et, .24f);
send(et, "hello event actor");
send(et, 42);
send(et, 24.f);
send(et, "hello event actor");
send(et, atom(":Exit"), exit_reason::user_defined);
await_all_others_done(); //auto et = spawn(new event_testee);
behavior_test<testee_behavior>();
behavior_test<event_testee>();
return CPPA_TEST_RESULT; return CPPA_TEST_RESULT;
......
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