Commit ecb9ab91 authored by Dominik Charousset's avatar Dominik Charousset

reimplemented delayed send, fixed logging

parent 5bd0f51d
......@@ -102,17 +102,9 @@ class behavior_stack_based : public Base {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
void become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
//FIXME
}
m_bhvr_stack.push_back(std::move(bhvr), mf);
}
virtual void become_waiting_for(behavior bhvr, message_id mf) = 0;
void do_become(behavior&& bhvr, bool discard_old) {
if (discard_old) m_bhvr_stack.pop_async_back();
m_bhvr_stack.push_back(std::move(bhvr));
}
virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
......@@ -135,6 +127,10 @@ class behavior_stack_based : public Base {
return m_bhvr_stack.sync_handler(msg_id);
}
inline void remove_handler(message_id mid) {
m_bhvr_stack.erase(mid);
}
protected:
// allows actors to keep previous behaviors and enables unbecome()
......
......@@ -307,6 +307,9 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
return none;
}
// unused in blocking actors
inline void remove_handler(message_id) { }
inline void dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
......@@ -336,12 +339,6 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
/*
void dequeue_response(behavior& bhvr, message_id request_id) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
}
*/
};
} // namespace cppa
......
......@@ -98,8 +98,6 @@ class proper_actor : public proper_actor_base<Base, SchedulingPolicy,
template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; }
inline void launch() {
auto bhvr = this->make_behavior();
if (bhvr) this->bhvr_stack().push_back(std::move(bhvr));
......@@ -107,8 +105,30 @@ class proper_actor : public proper_actor_base<Base, SchedulingPolicy,
}
bool invoke(mailbox_element* msg) {
return this->m_invoke_policy.invoke(this, msg, bhvr_stack().back(),
bhvr_stack().back_id());
return this->m_invoke_policy.invoke(this, msg, this->bhvr_stack().back(),
this->bhvr_stack().back_id());
}
void become_waiting_for(behavior bhvr, message_id mf) override {
if (bhvr.timeout().valid()) {
if (bhvr.timeout().valid()) {
this->invoke_policy().reset_timeout();
this->invoke_policy().request_timeout(this, bhvr.timeout());
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
void do_become(behavior&& bhvr, bool discard_old) override {
//if (discard_old) m_bhvr_stack.pop_async_back();
//m_bhvr_stack.push_back(std::move(bhvr));
this->invoke_policy().reset_timeout();
if (bhvr.timeout().valid()) {
this->invoke_policy().request_timeout(this, bhvr.timeout());
}
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
}
};
......
......@@ -374,8 +374,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
return mailbox_element::create(std::forward<Ts>(args)...);
}
void remove_handler(message_id id);
void cleanup(std::uint32_t reason);
// true if this actor receives EXIT messages as ordinary messages
......@@ -461,19 +459,6 @@ inline void local_actor::mark_arrived(message_id response_id) {
if (i != last) m_pending_responses.erase(i);
}
//inline detail::behavior_stack& local_actor::bhvr_stack() {
// return m_bhvr_stack;
//}
//inline void local_actor::do_become(const behavior& bhvr, bool discard_old) {
// behavior copy{bhvr};
// do_become(std::move(copy), discard_old);
//}
//inline void local_actor::remove_handler(message_id id) {
// m_bhvr_stack.erase(id);
//}
inline std::uint32_t local_actor::planned_exit_reason() const {
return m_planned_exit_reason;
}
......
......@@ -162,7 +162,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#define CPPA_LVL_NAME4() "TRACE"
#ifndef CPPA_LOG_LEVEL
# define CPPA_LOG_IMPL(lvlname, classname, funname, unused, message) { \
# define CPPA_LOG_IMPL(lvlname, classname, funname, message) { \
std::cerr << "[" << lvlname << "] " << classname << "::" \
<< funname << ": " << message << "\nStack trace:\n"; \
void *array[10]; \
......
......@@ -862,7 +862,7 @@ struct match_expr_from_type_list<util::type_list<Ts...> > {
template<typename... Lhs, typename... Rhs>
inline match_expr<Lhs..., Rhs...> operator,(const match_expr<Lhs...>& lhs,
const match_expr<Rhs...>& rhs) {
const match_expr<Rhs...>& rhs) {
return lhs.or_else(rhs);
}
......
......@@ -145,6 +145,12 @@ partial_function operator,(const match_expr<Cases...>& mexpr,
return mexpr.as_behavior_impl()->or_else(pfun.as_behavior_impl());
}
template<typename... Cases>
partial_function operator,(const partial_function& pfun,
const match_expr<Cases...>& mexpr) {
return pfun.as_behavior_impl()->or_else(mexpr.as_behavior_impl());
}
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
......
......@@ -262,9 +262,8 @@ class invoke_policy {
}
else if (v0 == atom("SYNC_TOUT")) {
CPPA_REQUIRE(!mid.valid());
//FIXME
//return self->waits_for_timeout(v1) ? timeout_message
// : expired_timeout_message;
return dptr()->waits_for_timeout(v1) ? timeout_message
: expired_timeout_message;
}
}
else if ( msg.size() == 1
......@@ -301,6 +300,8 @@ class invoke_policy {
Fun& fun,
MaybeResponseHandle hdl = MaybeResponseHandle{}) {
auto res = fun(msg); // might change mid
CPPA_LOG_DEBUG_IF(res, "actor did consume message");
CPPA_LOG_DEBUG_IF(!res, "actor did ignore message");
if (res) {
//message_header hdr{self, sender, mid.is_request() ? mid.response_id()
// : message_id{}};
......@@ -382,7 +383,8 @@ class invoke_policy {
Fun& fun,
message_id awaited_response) {
bool handle_sync_failure_on_mismatch = true;
if (dptr()->hm_should_skip(node)) { return hm_skip_msg;
if (dptr()->hm_should_skip(node)) {
return hm_skip_msg;
}
switch (this->filter_msg(self, node)) {
default: {
......@@ -412,7 +414,7 @@ class invoke_policy {
dptr()->handle_timeout(self, fun);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
//FIXME: self->remove_handler(awaited_response);
self->remove_handler(awaited_response);
}
return hm_msg_handled;
}
......@@ -437,7 +439,7 @@ class invoke_policy {
self->handle_sync_failure();
}
self->mark_arrived(awaited_response);
//FIXME: self->remove_handler(awaited_response);
self->remove_handler(awaited_response);
dptr()->hm_cleanup(self, previous_node);
return hm_msg_handled;
}
......@@ -459,6 +461,9 @@ class invoke_policy {
// no match (restore self members)
dptr()->hm_revert(self, previous_node);
}
CPPA_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: "
<< awaited_response.integer_value());
return hm_cache_msg;
}
}
......
......@@ -130,7 +130,13 @@ class no_scheduling {
});
util::fiber fself;
for (;;) {
await_data(self);
try { await_data(self); }
catch (std::exception& e) {
std::cerr << detail::demangle(typeid(e)) << ", what: "
<< e.what() << std::endl;
throw;
}
if (self->resume(&fself) == resumable::done) {
CPPA_LOG_DEBUG("resume returned resumable::done");
self->planned_exit_reason(exit_reason::normal);
......
......@@ -112,7 +112,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
//TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup));
delayed_send_helper().enqueue(message_header{}, std::move(tup));
}
template<typename Duration, typename... Data>
......@@ -124,7 +124,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
//TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup));
delayed_send_helper().enqueue(message_header{}, std::move(tup));
}
private:
......
......@@ -42,6 +42,8 @@ class scoped_actor {
scoped_actor();
explicit scoped_actor(bool hidden);
~scoped_actor();
inline blocking_untyped_actor* operator->() const {
......@@ -70,6 +72,9 @@ class scoped_actor {
private:
void init(bool hidden);
bool m_hidden;
actor_id m_prev;
intrusive_ptr<blocking_untyped_actor> m_self;
......
......@@ -64,7 +64,7 @@ const std::type_info* empty_tuple::type_token() const {
}
const std::string* empty_tuple::tuple_type_names() const {
static std::string result = "";
static std::string result = "@<>";
return &result;
}
......
......@@ -153,12 +153,10 @@ void local_actor::send_exit(const actor_addr& whom, std::uint32_t reason) {
send(detail::raw_access::get(whom), atom("EXIT"), reason);
}
void local_actor::remove_handler(message_id) {
}
void local_actor::delayed_send_tuple(const channel&, const util::duration&, cppa::any_tuple) {
void local_actor::delayed_send_tuple(const channel& dest,
const util::duration& rel_time,
cppa::any_tuple msg) {
get_scheduler()->delayed_send({address(), dest}, rel_time, std::move(msg));
}
response_promise local_actor::make_response_promise() {
......
......@@ -89,6 +89,8 @@ class scheduler_helper {
public:
scheduler_helper() : m_timer(true), m_printer(true) { }
void start() {
// launch threads
m_timer_thread = std::thread{&scheduler_helper::timer_loop, m_timer.get()};
......
......@@ -54,14 +54,29 @@ blocking_untyped_actor* alloc() {
} // namespace <anonymous>
scoped_actor::scoped_actor() : m_self(alloc()) {
m_prev = CPPA_SET_AID(m_self->id());
get_actor_registry()->inc_running();
void scoped_actor::init(bool hidden) {
m_hidden = hidden;
m_self.reset(alloc());
if (!hidden) {
m_prev = CPPA_SET_AID(m_self->id());
get_actor_registry()->inc_running();
}
}
scoped_actor::scoped_actor() {
init(false);
}
scoped_actor::scoped_actor(bool hidden) {
init(hidden);
}
scoped_actor::~scoped_actor() {
get_actor_registry()->dec_running();
CPPA_SET_AID(m_prev);
if (!m_hidden) {
get_actor_registry()->dec_running();
CPPA_SET_AID(m_prev);
}
}
} // namespace cppa
......@@ -62,7 +62,7 @@ struct thread_pool_scheduler::worker {
worker(job_queue* jq, job_ptr dummy) : m_job_queue(jq), m_dummy(dummy) { }
void start() {
//m_thread = std::thread(&thread_pool_scheduler::worker_loop, this);
m_thread = std::thread(&thread_pool_scheduler::worker_loop, this);
}
worker(const worker&) = delete;
......
......@@ -271,8 +271,10 @@ void serialize_impl(const any_tuple& tup, serializer* sink) {
auto tname = tup.tuple_type_names();
auto uti = get_uniform_type_info_map()->by_uniform_name(tname ? *tname : detail::get_tuple_type_names(*tup.vals()));
if (uti == nullptr) {
std::string err = "could not get uniform type info for ";
std::string err = "could not get uniform type info for \"";
err += tname ? *tname : detail::get_tuple_type_names(*tup.vals());
err += "\"";
CPPA_LOGF_ERROR(err);
throw std::runtime_error(err);
}
sink->begin_object(uti);
......@@ -790,7 +792,7 @@ class utim_impl : public uniform_type_info_map {
result = find_name(m_builtin_types, name);
result = (result) ? result : find_name(m_user_types, name);
}
if (!result && name.compare(0, 4, "@<>+") == 0) {
if (!result && name.compare(0, 3, "@<>") == 0) {
// create tuple UTI on-the-fly
result = insert(create_unique<default_meta_tuple>(name));
}
......
......@@ -413,11 +413,13 @@ void test_or_else() {
self->await_all_other_actors_done();
});
CPPA_LOGF_INFO("run_testee: handle_a.or_else(handle_b).or_else(handle_c)");
run_testee(
spawn([=] {
return handle_a.or_else(handle_b).or_else(handle_c);
})
);
CPPA_LOGF_INFO("run_testee: handle_a.or_else(handle_b), on(\"c\") ...");
run_testee(
spawn([=] {
return (
......@@ -426,6 +428,7 @@ void test_or_else() {
);
})
);
CPPA_LOGF_INFO("run_testee: on(\"a\") ..., handle_b.or_else(handle_c)");
run_testee(
spawn([=] {
return (
......@@ -482,7 +485,6 @@ void test_simple_reply_response() {
void test_spawn() {
test_simple_reply_response();
test_serial_reply();
return;
test_or_else();
test_continuation();
......@@ -492,19 +494,16 @@ void test_spawn() {
spawn<slave>(m);
spawn<slave>(m);
self->send(m, atom("done"));
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
CPPA_PRINT("test self->send()");
self->send(self, 1, 2, 3, true);
self->receive(on(1, 2, 3, true) >> [] { });
self->send(self, any_tuple{});
self->send_tuple(self, any_tuple{});
self->receive(on() >> [] { });
CPPA_CHECKPOINT();
self->send(self, any_tuple{});
self->receive(on() >> [] { });
CPPA_PRINT("test self->receive with zero timeout");
self->receive (
others() >> CPPA_UNEXPECTED_MSG_CB(),
......@@ -524,7 +523,7 @@ void test_spawn() {
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
}
......@@ -540,7 +539,7 @@ void test_spawn() {
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
}
......@@ -557,7 +556,7 @@ void test_spawn() {
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
}
......@@ -568,12 +567,13 @@ void test_spawn() {
on("hello echo") >> [] { },
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
CPPA_PRINT("test delayed_send()");
self->delayed_send(self, chrono::seconds(1), 1, 2, 3);
self->receive(on(1, 2, 3) >> [] { });
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
CPPA_PRINT("test timeout");
......@@ -581,11 +581,11 @@ void test_spawn() {
CPPA_CHECKPOINT();
spawn(testee1);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
spawn_event_testee2();
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto cstk = spawn<chopstick>();
......@@ -596,7 +596,7 @@ void test_spawn() {
self->send(cstk, atom("break"));
}
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto st = spawn<fixed_stack>(10);
......@@ -626,7 +626,7 @@ void test_spawn() {
}
// terminate st
self->send_exit(st, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto sync_testee1 = spawn<blocking_api>([](blocking_untyped_actor* self) {
......@@ -664,7 +664,7 @@ void test_spawn() {
others() >> CPPA_UNEXPECTED_MSG_CB(),
after(chrono::seconds(0)) >> [] { }
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
CPPA_PRINT("test sync send");
......@@ -708,7 +708,7 @@ void test_spawn() {
CPPA_CHECK_EQUAL(self->last_sender(), sync_testee);
}
);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
self->sync_send(sync_testee, "!?").await(
......@@ -738,7 +738,7 @@ void test_spawn() {
auto poison_pill = make_any_tuple(atom("done"));
anon_send(joe, poison_pill);
anon_send(bob, poison_pill);
await_all_actors_done();
self->await_all_other_actors_done();
function<actor (const string&, const actor&)> spawn_next;
// it's safe to capture spawn_next as reference here, because
......@@ -763,7 +763,7 @@ void test_spawn() {
};
auto joe_the_second = spawn(kr34t0r, "Joe", invalid_actor);
self->send(joe_the_second, atom("done"));
await_all_actors_done();
self->await_all_other_actors_done();
auto f = [](const string& name) -> behavior {
return (
......@@ -788,7 +788,7 @@ void test_spawn() {
);
self->send_exit(a1, exit_reason::user_shutdown);
self->send_exit(a2, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
auto res1 = behavior_test<testee_actor>(spawn<blocking_api>(testee_actor{}));
CPPA_CHECK_EQUAL("wait4int", res1);
......@@ -804,7 +804,7 @@ void test_spawn() {
self->become(others() >> CPPA_UNEXPECTED_MSG_CB());
});
self->send_exit(legion, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
self->trap_exit(true);
auto ping_actor = self->spawn<monitored+blocking_api>(ping, 10);
......@@ -843,16 +843,16 @@ void test_spawn() {
}
);
// wait for termination of all spawned actors
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECK_EQUAL(flags, 0x0F);
// verify pong messages
CPPA_CHECK_EQUAL(pongs(), 10);
CPPA_CHECKPOINT();
spawn<priority_aware>(high_priority_testee);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
spawn<high_priority_testee_class, priority_aware>();
await_all_actors_done();
self->await_all_other_actors_done();
// don't try this at home, kids
self->send(self, atom("check"));
self->receive (
......
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