Commit 3eee072f authored by Dominik Charousset's avatar Dominik Charousset

documentation & implementations for new policies

parent b0dc9d4d
...@@ -150,7 +150,7 @@ set(LIBCPPA_SRC ...@@ -150,7 +150,7 @@ set(LIBCPPA_SRC
src/deserializer.cpp src/deserializer.cpp
src/duration.cpp src/duration.cpp
src/empty_tuple.cpp src/empty_tuple.cpp
src/event_based_actor.cpp src/event_based_resume.cpp
src/exception.cpp src/exception.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/fd_util.cpp src/fd_util.cpp
......
...@@ -331,3 +331,10 @@ unit_testing/test_uniform_type.cpp ...@@ -331,3 +331,10 @@ unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp unit_testing/test_yield_interface.cpp
cppa/detail/proper_actor.hpp cppa/detail/proper_actor.hpp
cppa/policy/no_resume.hpp cppa/policy/no_resume.hpp
cppa/policy.hpp
cppa/detail/functor_based_actor.hpp
cppa/policy/no_scheduling.hpp
cppa/policy/resume_policy.hpp
cppa/policy/cooperative_scheduling.hpp
cppa/policy/scheduling_policy.hpp
cppa/policy/priority_policy.hpp
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_FUNCTOR_BASED_ACTOR_HPP
#define CPPA_FUNCTOR_BASED_ACTOR_HPP
#include "cppa/untyped_actor.hpp"
namespace cppa { namespace detail {
class functor_based_actor : public untyped_actor {
};
} } // namespace cppa::detail
#endif // CPPA_FUNCTOR_BASED_ACTOR_HPP
#ifndef PROPER_ACTOR_HPP #ifndef PROPER_ACTOR_HPP
#define PROPER_ACTOR_HPP #define PROPER_ACTOR_HPP
#include <type_traits>
#include "cppa/blocking_actor.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
namespace cppa { namespace util { class fiber; } } namespace cppa { namespace util { class fiber; } }
...@@ -11,7 +14,8 @@ template<class Base, ...@@ -11,7 +14,8 @@ template<class Base,
class SchedulingPolicy, class SchedulingPolicy,
class PriorityPolicy, class PriorityPolicy,
class ResumePolicy, class ResumePolicy,
class InvokePolicy> class InvokePolicy,
bool OverrideDequeue = std::is_base_of<blocking_actor, Base>::value>
class proper_actor : public Base { class proper_actor : public Base {
public: public:
...@@ -28,18 +32,23 @@ class proper_actor : public Base { ...@@ -28,18 +32,23 @@ class proper_actor : public Base {
} }
inline void launch() { inline void launch() {
m_resume_policy.launch(this); m_scheduling_policy.launch(this);
} }
inline mailbox_element* next_message() { inline mailbox_element* next_message() {
m_priority_policy.next_message(this); return m_priority_policy.next_message(this);
} }
void invoke_message(mailbox_element* msg) override { inline void invoke(mailbox_element* msg) {
m_invoke_policy.invoke(this, msg); m_invoke_policy.invoke(this, msg);
} }
private: // grant access to the actor's mailbox
Base::mailbox_type& mailbox() {
return this->m_mailbox;
}
protected:
SchedulingPolicy m_scheduling_policy; SchedulingPolicy m_scheduling_policy;
PriorityPolicy m_priority_policy; PriorityPolicy m_priority_policy;
...@@ -48,6 +57,61 @@ class proper_actor : public Base { ...@@ -48,6 +57,61 @@ class proper_actor : public Base {
}; };
} // namespace cppa::detail // for blocking actors, there's one more member function to implement
template<class Base,
class SchedulingPolicy,
class PriorityPolicy,
class ResumePolicy,
class InvokePolicy>
class proper_actor<Base,
SchedulingPolicy,
PriorityPolicy,
ResumePolicy,
InvokePolicy,
true> : public proper_actor<Base,
SchedulingPolicy,
PriorityPolicy,
ResumePolicy,
InvokePolicy,
false> {
public:
void dequeue(behavior& bhvr) override {
if (bhvr.timeout().valid()) {
auto abs_time = m_scheduling_policy.init_timeout(this, bhvr.timeout());
auto done = false;
while (!done) {
if (!m_resume_policy.await_data(this, abs_time)) {
bhvr.handle_timeout();
done = true;
}
else {
auto msg = m_priority_policy.next_message(this);
// must not return nullptr, because await_data guarantees
// at least one message in our mailbox
CPPA_REQUIRE(msg != nullptr);
done = m_invoke_policy.invoke(this, bhvr, msg);
}
}
}
else {
for (;;) {
auto msg = m_priority_policy.next_message(this);
while (msg) {
if (m_invoke_policy.invoke(this, bhvr, msg)) {
// we're done
return;
}
msg = m_priority_policy.next_message(this);
}
m_resume_policy.await_data(this);
}
}
}
};
} } // namespace cppa::detail
#endif // PROPER_ACTOR_HPP #endif // PROPER_ACTOR_HPP
...@@ -400,16 +400,6 @@ typedef intrusive_ptr<local_actor> local_actor_ptr; ...@@ -400,16 +400,6 @@ typedef intrusive_ptr<local_actor> local_actor_ptr;
/** @cond PRIVATE */ /** @cond PRIVATE */
inline void local_actor::dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
}
inline void local_actor::dequeue_response(behavior&& bhvr, message_id id) {
behavior tmp{std::move(bhvr)};
dequeue_response(tmp, id);
}
inline bool local_actor::trap_exit() const { inline bool local_actor::trap_exit() const {
return m_trap_exit; return m_trap_exit;
} }
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_POLICY_HPP
#define CPPA_POLICY_HPP
#include "cppa/policy/no_resume.hpp"
#include "cppa/policy/prioritizing.hpp"
#include "cppa/policy/nestable_invoke.hpp"
#include "cppa/policy/sequential_invoke.hpp"
#include "cppa/policy/event_based_resume.hpp"
#include "cppa/policy/invoke_policy_base.hpp"
#include "cppa/policy/context_switching_resume.hpp"
#endif // CPPA_POLICY_HPP
...@@ -34,6 +34,11 @@ ...@@ -34,6 +34,11 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/detail/yield_interface.hpp"
namespace cppa { class local_actor; }
namespace cppa { namespace policy { namespace cppa { namespace policy {
/** /**
...@@ -44,26 +49,79 @@ class context_switching_resume { ...@@ -44,26 +49,79 @@ class context_switching_resume {
public: public:
/**
* @brief Creates a context-switching actor running @p fun.
*/
//context_switching_resume(std::function<void()> fun);
resume_result resume(abstract_actor*, util::fiber* from);
typedef std::chrono::high_resolution_clock::time_point timeout_type;
mailbox_element* next_message(); template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) {
inline mailbox_element* next_message(int) { auto e = self->m_mailbox.try_pop();
// we don't use the dummy element returned by init_timeout while (e == nullptr) {
return next_message(); if (self->m_mailbox.can_fetch_more() == false) {
self->set_state(actor_state::about_to_block);
// make sure mailbox is empty
if (self->m_mailbox.can_fetch_more()) {
// someone preempt us => continue
self->set_state(actor_state::ready);
}
// wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked);
}
}
// ok, we have at least one message
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
}
} }
int init_timeout(const util::duration& rel_time); template<class Actor, typename F>
void try_fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop();
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
}
}
inline mailbox_element* try_pop() { template<class Actor>
return m_mailbox.try_pop(); resume_result resume(Actor* self, util::fiber* from) {
CPPA_LOG_TRACE("state = " << static_cast<int>(self->state()));
CPPA_REQUIRE(from != nullptr);
CPPA_REQUIRE(next_job == nullptr);
using namespace detail;
for (;;) {
switch (call(&m_fiber, from)) {
case yield_state::done: {
CPPA_REQUIRE(next_job == nullptr);
return resume_result::actor_done;
}
case yield_state::ready: {
break;
}
case yield_state::blocked: {
CPPA_REQUIRE(next_job == nullptr);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
CPPA_REQUIRE(next_job == nullptr);
break;
}
case actor_state::blocked: {
// wait until someone re-schedules that actor
return resume_result::actor_blocked;
}
default: {
CPPA_CRITICAL("illegal yield result");
}
}
break;
}
default: {
CPPA_CRITICAL("illegal state");
}
}
}
} }
private: private:
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_COOPERATIVE_SCHEDULING_HPP
#define CPPA_COOPERATIVE_SCHEDULING_HPP
namespace cppa { namespace policy {
class cooperative_scheduling {
public:
using timeout_type = int;
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
// this does return nullptr
template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop();
while (e == nullptr) {
if (self->m_mailbox.can_fetch_more() == false) {
self->set_state(actor_state::about_to_block);
// make sure mailbox is empty
if (self->m_mailbox.can_fetch_more()) {
// someone preempt us => continue
self->set_state(actor_state::ready);
}
// wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked);
}
}
// ok, we have at least one message
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
}
}
template<class Actor, typename F>
inline void fetch_messages(Actor* self, F cb, timeout_type) {
// a call to this call is always preceded by init_timeout,
// which will trigger a timeout message
fetch_messages(self, cb);
}
};
} } // namespace cppa::policy
#endif // CPPA_COOPERATIVE_SCHEDULING_HPP
...@@ -35,13 +35,12 @@ ...@@ -35,13 +35,12 @@
#include <stack> #include <stack>
#include <memory> #include <memory>
#include <vector> #include <vector>
#include <type_traits>
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/stackless.hpp"
#include "cppa/scheduled_actor.hpp" #include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
namespace cppa { namespace policy { namespace cppa { namespace policy {
...@@ -51,32 +50,154 @@ namespace cppa { namespace policy { ...@@ -51,32 +50,154 @@ namespace cppa { namespace policy {
*/ */
class event_based_resume { class event_based_resume {
friend class detail::receive_policy;
typedef combined_type super;
public: public:
resume_result resume(untyped_actor*, util::fiber*); template<class Actor>
void await_data(Actor*) {
/** static_assert(std::is_same<Actor, Actor>::value == false,
* @brief Initializes the actor. "The event_based_resume policy cannot be used "
*/ "to implement blocking actors");
virtual behavior make_behavior() = 0; }
scheduled_actor_type impl_type(); template<class Actor>
bool await_data(Actor*, const util::duration&) {
static intrusive_ptr<event_based_actor> from(std::function<void()> fun); static_assert(std::is_same<Actor, Actor>::value == false,
"The event_based_resume policy cannot be used "
static intrusive_ptr<event_based_actor> from(std::function<behavior()> fun); "to implement blocking actors");
}
static intrusive_ptr<event_based_actor> from(std::function<void(event_based_actor*)> fun);
template<class Actor>
static intrusive_ptr<event_based_actor> from(std::function<behavior(event_based_actor*)> fun); timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
protected: self->request_timeout(rel_time);
return 0; // return some dummy value
event_based_actor(actor_state st = actor_state::blocked); }
template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop();
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
}
}
template<class Actor, typename F>
inline void fetch_messages(Actor* self, F cb, timeout_type) {
// a call to this call is always preceded by init_timeout,
// which will trigger a timeout message
fetch_messages(self, cb);
}
template<class Actor, typename F>
void try_fetch_messages(Actor* self, F cb) {
// try_fetch and fetch have the same semantics for event-based actors
fetch_messages(self, cb);
}
template<class Actor>
resume_result resume(Actor* self, util::fiber*) {
CPPA_LOG_TRACE("id = " << self->id()
<< ", state = " << static_cast<int>(self->state()));
CPPA_REQUIRE( self->state() == actor_state::ready
|| self->state() == actor_state::pending);
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (self->exit_reason() == self->exit_reason::not_exited) {
if (self->planned_exit_reason() == self->exit_reason::not_exited) {
self->planned_exit_reason(exit_reason::normal);
}
self->on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
cleanup(planned_exit_reason());
}
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
CPPA_REQUIRE(next_job == nullptr);
return true;
};
CPPA_REQUIRE(next_job == nullptr);
try {
//auto e = m_mailbox.try_pop();
for (auto e = m_mailbox.try_pop(); ; e = m_mailbox.try_pop()) {
//e = m_mailbox.try_pop();
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) {
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready:
// interrupted by arriving message
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
break;
case actor_state::blocked:
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked");
// done setting actor to blocked
return resume_result::actor_blocked;
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid state");
CPPA_CRITICAL("invalid state");
};
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"mailbox can fetch more");
set_state(actor_state::ready);
CPPA_REQUIRE(m_chained_actor == nullptr);
}
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "try to invoke message: " << to_string(e->msg));
if (m_bhvr_stack.invoke(m_recv_policy, this, e)) {
CPPA_LOG_DEBUG_IF(m_chained_actor,
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (m_bhvr_stack.empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
return resume_result::actor_done;
}
m_bhvr_stack.cleanup();
}
}
}
}
catch (actor_exited& what) {
CPPA_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason());
if (exit_reason() == exit_reason::not_exited) {
quit(what.reason());
}
}
catch (std::exception& e) {
CPPA_LOG_WARNING("actor died because of exception: "
<< detail::demangle(typeid(e))
<< ", what() = " << e.what());
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
catch (...) {
CPPA_LOG_WARNING("actor died because of an unknown exception");
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
done_cb();
return resume_result::actor_done;
}
}; };
......
...@@ -62,12 +62,13 @@ enum receive_policy_flag { ...@@ -62,12 +62,13 @@ enum receive_policy_flag {
template<receive_policy_flag X> template<receive_policy_flag X>
struct rp_flag { typedef std::integral_constant<receive_policy_flag, X> type; }; struct rp_flag { typedef std::integral_constant<receive_policy_flag, X> type; };
template<class Derived>
class invoke_policy_base { class invoke_policy_base {
public: public:
typedef mailbox_element* pointer; typedef mailbox_element* pointer;
typedef std::unique_ptr<mailbox_element, disposer> smart_pointer; typedef std::unique_ptr<mailbox_element, detail::disposer> smart_pointer;
enum handle_message_result { enum handle_message_result {
hm_timeout_msg, hm_timeout_msg,
...@@ -213,7 +214,7 @@ class invoke_policy_base { ...@@ -213,7 +214,7 @@ class invoke_policy_base {
private: private:
std::list<std::unique_ptr<mailbox_element, disposer> > m_cache; std::list<std::unique_ptr<mailbox_element, detail::disposer> > m_cache;
template<class Client> template<class Client>
inline void handle_timeout(Client* client, behavior& bhvr) { inline void handle_timeout(Client* client, behavior& bhvr) {
...@@ -316,7 +317,7 @@ class invoke_policy_base { ...@@ -316,7 +317,7 @@ class invoke_policy_base {
if (fhdl.valid()) fhdl.apply(make_any_tuple(atom("VOID"))); if (fhdl.valid()) fhdl.apply(make_any_tuple(atom("VOID")));
} }
} else { } else {
if ( matches<atom_value, std::uint64_t>(*res) if ( detail::matches<atom_value, std::uint64_t>(*res)
&& res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) { && res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) {
auto id = res->template get_as<std::uint64_t>(1); auto id = res->template get_as<std::uint64_t>(1);
auto msg_id = message_id::from_integer_value(id); auto msg_id = message_id::from_integer_value(id);
...@@ -459,6 +460,6 @@ class invoke_policy_base { ...@@ -459,6 +460,6 @@ class invoke_policy_base {
}; };
} } // namespace cppa::detail } } // namespace cppa::policy
#endif // CPPA_NESTABLE_RECEIVE_POLICY_HPP #endif // CPPA_NESTABLE_RECEIVE_POLICY_HPP
#ifndef NO_RESUME_HPP #ifndef NO_RESUME_HPP
#define NO_RESUME_HPP #define NO_RESUME_HPP
#include <chrono>
#include "cppa/policy/resume_policy.hpp"
namespace cppa { namespace policy { namespace cppa { namespace policy {
class no_resume { class no_resume {
public:
using timeout_type = std::chrono::high_resolution_clock::time_point;
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop();
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
}
}
template<class Actor, typename F>
inline void fetch_messages(Actor* self, F cb, timeout_type) {
// a call to this call is always preceded by init_timeout,
// which will trigger a timeout message
fetch_messages(self, cb);
}
template<class Actor, typename F>
void try_fetch_messages(Actor* self, F cb) {
// try_fetch and fetch have the same semantics for event-based actors
fetch_messages(self, cb);
}
template<class Actor>
resume_result resume(Actor* self, util::fiber*) {
CPPA_LOG_TRACE("id = " << self->id()
<< ", state = " << static_cast<int>(self->state()));
CPPA_REQUIRE( self->state() == actor_state::ready
|| self->state() == actor_state::pending);
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (self->exit_reason() == self->exit_reason::not_exited) {
if (self->planned_exit_reason() == self->exit_reason::not_exited) {
self->planned_exit_reason(exit_reason::normal);
}
self->on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
cleanup(planned_exit_reason());
}
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
CPPA_REQUIRE(next_job == nullptr);
return true;
};
CPPA_REQUIRE(next_job == nullptr);
try {
//auto e = m_mailbox.try_pop();
for (auto e = m_mailbox.try_pop(); ; e = m_mailbox.try_pop()) {
//e = m_mailbox.try_pop();
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) {
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready:
// interrupted by arriving message
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
break;
case actor_state::blocked:
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked");
// done setting actor to blocked
return resume_result::actor_blocked;
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid state");
CPPA_CRITICAL("invalid state");
};
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"mailbox can fetch more");
set_state(actor_state::ready);
CPPA_REQUIRE(m_chained_actor == nullptr);
}
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "try to invoke message: " << to_string(e->msg));
if (m_bhvr_stack.invoke(m_recv_policy, this, e)) {
CPPA_LOG_DEBUG_IF(m_chained_actor,
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (m_bhvr_stack.empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
return resume_result::actor_done;
}
m_bhvr_stack.cleanup();
}
}
}
}
catch (actor_exited& what) {
CPPA_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason());
if (exit_reason() == exit_reason::not_exited) {
quit(what.reason());
}
}
catch (std::exception& e) {
CPPA_LOG_WARNING("actor died because of exception: "
<< detail::demangle(typeid(e))
<< ", what() = " << e.what());
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
catch (...) {
CPPA_LOG_WARNING("actor died because of an unknown exception");
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
done_cb();
return resume_result::actor_done;
}
}; };
} // namespace cppa::policy } } // namespace cppa::policy
#endif // NO_RESUME_HPP #endif // NO_RESUME_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_NO_SCHEDULING_HPP
#define CPPA_NO_SCHEDULING_HPP
#include <mutex>
#include <thread>
#include <condition_variable>
#include "cppa/detail/sync_request_bouncer.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace policy {
class no_scheduling {
typedef std::unique_lock<std::mutex> lock_type;
public:
template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
auto ptr = self->new_mailbox_element(hdr, std::move(msg));
switch (self->mailbox().enqueue(ptr)) {
case intrusive::first_enqueued: {
lock_type guard(m_mtx);
m_cv.notify_one();
break;
}
default: break;
case intrusive::queue_closed:
if (hdr.id.valid()) {
detail::sync_request_bouncer f{self->exit_reason()};
f(hdr.sender, hdr.id);
}
break;
}
}
template<class Actor>
void launch(Actor* self) {
std::thread([=] {
auto rr = resume_result::actor_blocked;
while (rr != resume_result::actor_done) {
wait_for_data();
self->resume();
}
}).detach();
}
private:
bool timed_wait_for_data(const timeout_type& abs_time) {
CPPA_REQUIRE(not this->m_mailbox.closed());
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (mailbox_empty()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) {
return false;
}
}
}
return true;
}
void wait_for_data() {
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (mailbox_empty()) m_cv.wait(guard);
}
}
std::mutex m_mtx;
std::condition_variable m_cv;
};
} } // namespace cppa::policy
#endif // CPPA_NO_SCHEDULING_HPP
...@@ -37,13 +37,19 @@ ...@@ -37,13 +37,19 @@
#include "cppa/message_priority.hpp" #include "cppa/message_priority.hpp"
#include "cppa/detail/sync_request_bouncer.hpp" #include "cppa/detail/sync_request_bouncer.hpp"
namespace cppa { namespace cppa { namespace policy {
class prioritizing { class prioritizing {
public: public:
mailbox_element* try_pop() override { template<typename Actor>
mailbox_element* next_message(Actor* self) {
return self->m_mailbox.try_pop();
}
/*
mailbox_element* try_pop() {
auto result = m_high_priority_mailbox.try_pop(); auto result = m_high_priority_mailbox.try_pop();
return (result) ? result : this->m_mailbox.try_pop(); return (result) ? result : this->m_mailbox.try_pop();
} }
...@@ -55,18 +61,18 @@ class prioritizing { ...@@ -55,18 +61,18 @@ class prioritizing {
typedef prioritizing combined_type; typedef prioritizing combined_type;
void cleanup(std::uint32_t reason) override { void cleanup(std::uint32_t reason) {
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{reason};
m_high_priority_mailbox.close(f); m_high_priority_mailbox.close(f);
Base::cleanup(reason); Base::cleanup(reason);
} }
bool mailbox_empty() override { bool mailbox_empty() {
return m_high_priority_mailbox.empty() return m_high_priority_mailbox.empty()
&& this->m_mailbox.empty(); && this->m_mailbox.empty();
} }
void enqueue(const message_header& hdr, any_tuple msg) override { void enqueue(const message_header& hdr, any_tuple msg) {
typename Base::mailbox_type* mbox = nullptr; typename Base::mailbox_type* mbox = nullptr;
if (hdr.priority == message_priority::high) { if (hdr.priority == message_priority::high) {
mbox = &m_high_priority_mailbox; mbox = &m_high_priority_mailbox;
...@@ -78,9 +84,10 @@ class prioritizing { ...@@ -78,9 +84,10 @@ class prioritizing {
} }
typename Base::mailbox_type m_high_priority_mailbox; typename Base::mailbox_type m_high_priority_mailbox;
*/
}; };
} // namespace cppa } } // namespace cppa::policy
#endif // PRIORITIZING_HPP #endif // PRIORITIZING_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_PRIORITY_POLICY_HPP
#define CPPA_PRIORITY_POLICY_HPP
namespace cppa { class mailbox_element; }
namespace cppa { namespace policy {
/**
* @brief The priority_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all
* required member function and their behavior for any priority policy.
*/
class priority_policy {
public:
/**
* @brief Returns the next message from the list of cached elements or
* @p nullptr. The latter indicates only that there is no element
* left in the cache.
*/
template<class Actor>
mailbox_element* next_message(Actor* self);
/**
* @brief Fetches new messages from the actor's mailbox. The member
* function returns @p false if no message was read,
* @p true otherwise.
*
* This member function calls {@link scheduling_policy::fetch_messages},
* so a returned @p false value indicates that the client must prepare
* for re-scheduling.
*/
template<class Actor, typename F>
bool fetch_messages(Actor* self);
};
} } // namespace cppa::policy
#endif // CPPA_PRIORITY_POLICY_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_RESUME_POLICY_HPP
#define CPPA_RESUME_POLICY_HPP
// this header consists all type definitions needed to
// implement the resume_policy trait
namespace cppa { namespace util { class duration; } }
namespace cppa { namespace policy {
enum class resume_result {
actor_blocked,
actor_done
};
/**
* @brief The resume_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all
* required member function and their behavior for any resume policy.
*/
class resume_policy {
public:
/**
* @brief Resumes the actor by reading a new message <tt>msg</tt> and then
* calling <tt>self->invoke(msg)</tt>. This process is repeated
* until either no message is left in the actor's mailbox or the
* actor finishes execution.
*/
template<class Actor>
resume_result resume(Actor* self, util::fiber* from);
/**
* @brief Waits unconditionally until a new message arrives.
* @note This member function must raise a compiler error if the resume
* strategy cannot be used to implement blocking actors.
*/
template<class Actor>
bool await_data(Actor* self);
/**
* @brief Waits until either a new message arrives, or a timeout occurs.
* The @p abs_time argument is the return value of
* {@link scheduling_policy::init_timeout}. Returns true if a
* message arrived in time, otherwise false.
* @note This member function must raise a compiler error if the resume
* strategy cannot be used to implement blocking actors.
*/
template<class Actor, typename AbsTimeout>
bool await_data(Actor* self, const AbsTimeout& abs_time);
};
} } // namespace cppa::policy
#endif // CPPA_RESUME_POLICY_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_SCHEDULING_POLICY_HPP
#define CPPA_SCHEDULING_POLICY_HPP
namespace cppa { class message_header; class any_tuple; }
namespace cppa { namespace util { class duration; } }
namespace cppa { namespace policy {
enum class timed_fetch_result {
no_message,
indeterminate,
success
};
/**
* @brief The scheduling_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all
* required member function and their behavior for any scheduling policy.
*/
class scheduling_policy {
public:
/**
* @brief This typedef can be set freely by any implementation and is
* used by callers to pass the result of @p init_timeout back to
* @p fetch_messages.
*/
using timeout_type = int;
/**
* @brief Requests a timeout for the next call to @p fetch_message.
*/
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time);
/**
* @brief Fetches new messages from the actor's mailbox and feeds them
* to the given callback. The member function returns @p false if
* no message was read, @p true otherwise.
*
* In case this function returned @p false, the policy also sets the state
* of the actor to blocked. Any caller must evaluate the return value and
* act properly in case of a returned @p false, i.e., it must <b>not</b>
* atttempt to call any further function on the actor, since it might be
* already in the pipe for re-scheduling.
*/
template<class Actor, typename F>
bool fetch_messages(Actor* self, F cb);
/**
* @brief Tries to fetch new messages from the actor's mailbox and to feeds
* them to the given callback. The member function returns @p false
* if no message was read, @p true otherwise.
*
* This member function does not have any side-effect other than removing
* messages from the actor's mailbox.
*/
template<class Actor, typename F>
bool try_fetch_messages(Actor* self, F cb);
/**
* @brief Tries to fetch new messages before a timeout occurs. The message
* can either return @p success, or @p no_message,
* or @p indeterminate. The latter occurs for cooperative scheduled
* operations and means that timeouts are signaled using
* special-purpose messages. In this case, clients have to simply
* wait for the arriving message.
*/
template<class Actor, typename F>
timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time);
/**
* @brief Enqueues given message to the actor's mailbox and take any
* steps to resume the actor if it's currently blocked.
*/
template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg);
/**
* @brief Starts the given actor either by launching a thread or enqueuing
* it to the cooperative scheduler's job queue.
*/
template<class Actor>
void launch(Actor* self);
};
} } // namespace cppa::policy
#endif // CPPA_SCHEDULING_POLICY_HPP
...@@ -52,10 +52,7 @@ namespace cppa { ...@@ -52,10 +52,7 @@ namespace cppa {
class scheduler; class scheduler;
namespace util { struct fiber; } namespace util { struct fiber; }
enum class resume_result {
actor_blocked,
actor_done
};
enum scheduled_actor_type { enum scheduled_actor_type {
context_switching_impl, // enqueued to the job queue on startup context_switching_impl, // enqueued to the job queue on startup
......
...@@ -33,13 +33,12 @@ ...@@ -33,13 +33,12 @@
#include <type_traits> #include <type_traits>
#include "cppa/threaded.hpp" #include "cppa/policy.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/typed_actor.hpp" #include "cppa/typed_actor.hpp"
#include "cppa/prioritizing.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp" #include "cppa/detail/proper_actor.hpp"
namespace cppa { namespace cppa {
...@@ -56,22 +55,6 @@ constexpr bool unbound_spawn_options(spawn_options opts) { ...@@ -56,22 +55,6 @@ constexpr bool unbound_spawn_options(spawn_options opts) {
* @{ * @{
*/ */
/**
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
static_assert(unbound_spawn_options(Options),
"top-level spawns cannot have monitor or link flag");
return get_scheduler()->exec(Options,
scheduler::init_callback{},
std::forward<Ts>(args)...);
}
/** /**
* @brief Spawns an actor of type @p Impl. * @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments. * @param args Constructor arguments.
...@@ -85,6 +68,39 @@ actor spawn(Ts&&... args) { ...@@ -85,6 +68,39 @@ actor spawn(Ts&&... args) {
"Impl is not a derived type of event_based_actor"); "Impl is not a derived type of event_based_actor");
static_assert(unbound_spawn_options(Options), static_assert(unbound_spawn_options(Options),
"top-level spawns cannot have monitor or link flag"); "top-level spawns cannot have monitor or link flag");
static_assert(unbound_spawn_options(Options),
"top-level spawns cannot have monitor or link flag");
using scheduling_policy = typename std::conditional<
has_detach_flag(Options),
policy::no_scheduling,
policy::cooperative_scheduling
>::type;
using priority_policy = typename std::conditional<
has_priority_aware_flag(Options),
policy::prioritizing,
policy::not_prioritizing
>::type;
using resume_policy = typename std::conditional<
has_blocking_api_flag(Options),
typename std::conditional<
has_detach_flag(Options),
policy::no_resume,
policy::context_switching_resume
>::type,
policy::event_based_resume
>::type;
using invoke_policy = typename std::conditional<
has_blocking_api_flag(Options),
policy::nestable_invoke,
policy::sequential_invoke
>::type;
using proper_impl = detail::proper_actor<Impl,
scheduling_policy,
priority_policy,
resume_policy,
invoke_policy>;
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
/*
scheduled_actor_ptr ptr; scheduled_actor_ptr ptr;
if (has_priority_aware_flag(Options)) { if (has_priority_aware_flag(Options)) {
using derived = typename extend<Impl>::template with<threaded, prioritizing>; using derived = typename extend<Impl>::template with<threaded, prioritizing>;
...@@ -96,6 +112,30 @@ actor spawn(Ts&&... args) { ...@@ -96,6 +112,30 @@ actor spawn(Ts&&... args) {
} }
else ptr = make_counted<Impl>(std::forward<Ts>(args)...); else ptr = make_counted<Impl>(std::forward<Ts>(args)...);
return get_scheduler()->exec(Options, std::move(ptr)); return get_scheduler()->exec(Options, std::move(ptr));
*/
}
/**
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
using base_class = typename std::conditional<
has_blocking_api_flag(Options),
detail::functor_based_blocking_actor,
detail::functor_based_actor
>::type;
return spawn<base_class>(std::forward<Ts>(args)...);
using impl = detail::proper_actor<untyped_actor,
scheduling_policy,
priority_policy,
resume_policy,
invoke_policy>;
return make_counted<impl>();
} }
/** /**
...@@ -106,6 +146,7 @@ actor spawn(Ts&&... args) { ...@@ -106,6 +146,7 @@ actor spawn(Ts&&... args) {
* @returns An {@link actor} to the spawned {@link actor}. * @returns An {@link actor} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns. * @note The spawned has joined the group before this function returns.
*/ */
/*
template<spawn_options Options = no_spawn_options, typename... Ts> template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) { actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided"); static_assert(sizeof...(Ts) > 0, "too few arguments provided");
...@@ -117,6 +158,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) { ...@@ -117,6 +158,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
init_cb, init_cb,
std::forward<Ts>(args)...)); std::forward<Ts>(args)...));
} }
*/
/** /**
* @brief Spawns an actor of type @p Impl that immediately joins @p grp. * @brief Spawns an actor of type @p Impl that immediately joins @p grp.
...@@ -126,6 +168,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) { ...@@ -126,6 +168,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @returns An {@link actor} to the spawned {@link actor}. * @returns An {@link actor} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns. * @note The spawned has joined the group before this function returns.
*/ */
/*
template<class Impl, spawn_options Options, typename... Ts> template<class Impl, spawn_options Options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) { actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...); auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
...@@ -144,6 +187,7 @@ typename Impl::typed_pointer_type spawn_typed(Ts&&... args) { ...@@ -144,6 +187,7 @@ typename Impl::typed_pointer_type spawn_typed(Ts&&... args) {
eval_sopts(Options, get_scheduler()->exec(Options, std::move(p))) eval_sopts(Options, get_scheduler()->exec(Options, std::move(p)))
); );
} }
*/
/*TODO: /*TODO:
template<spawn_options Options, typename... Ts> template<spawn_options Options, typename... Ts>
......
...@@ -38,45 +38,8 @@ ...@@ -38,45 +38,8 @@
namespace cppa { namespace policy { namespace cppa { namespace policy {
/*
context_switching_resume::context_switching_resume(std::function<void()> fun)
: super(actor_state::ready, true)
, m_fiber(&context_switching_actor::trampoline, this) {
set_behavior(std::move(fun));
}
*/
int context_switching_actor::init_timeout(const util::duration& tout) {
// request explicit timeout message
request_timeout(tout);
return 0;
}
mailbox_element* context_switching_actor::next_message(const timeout_type&) {
return await_message();
}
mailbox_element* context_switching_actor::next_message() {
auto e = m_mailbox.try_pop();
while (e == nullptr) {
if (m_mailbox.can_fetch_more() == false) {
set_state(actor_state::about_to_block);
// make sure mailbox is empty
if (m_mailbox.can_fetch_more()) {
// someone preempt us => continue
set_state(actor_state::ready);
}
// wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked);
}
e = m_mailbox.try_pop();
}
return e;
}
void context_switching_resume::trampoline(void* this_ptr) { void context_switching_resume::trampoline(void* this_ptr) {
auto _this = reinterpret_cast<context_switching_actor*>(this_ptr); auto _this = reinterpret_cast<context_switching_resume*>(this_ptr);
bool cleanup_called = false; bool cleanup_called = false;
try { _this->run(); } try { _this->run(); }
catch (actor_exited&) { catch (actor_exited&) {
...@@ -93,51 +56,6 @@ void context_switching_resume::trampoline(void* this_ptr) { ...@@ -93,51 +56,6 @@ void context_switching_resume::trampoline(void* this_ptr) {
detail::yield(detail::yield_state::done); detail::yield(detail::yield_state::done);
} }
resume_result context_switching_resume::resume(util::fiber* from) {
CPPA_LOGMF(CPPA_TRACE, this, "state = " << static_cast<int>(state()));
CPPA_REQUIRE(from != nullptr);
CPPA_REQUIRE(next_job == nullptr);
using namespace detail;
scoped_self_setter sss{this};
for (;;) {
switch (call(&m_fiber, from)) {
case yield_state::done: {
CPPA_REQUIRE(next_job == nullptr);
m_chained_actor.swap(next_job);
return resume_result::actor_done;
}
case yield_state::ready: {
break;
}
case yield_state::blocked: {
CPPA_REQUIRE(next_job == nullptr);
m_chained_actor.swap(next_job);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
m_chained_actor.swap(next_job);
CPPA_REQUIRE(next_job == nullptr);
break;
}
case actor_state::blocked: {
// wait until someone re-schedules that actor
return resume_result::actor_blocked;
}
default: {
CPPA_CRITICAL("illegal yield result");
}
}
break;
}
default: {
CPPA_CRITICAL("illegal state");
}
}
}
}
} } // namespace cppa::policy } } // namespace cppa::policy
#else // ifdef CPPA_DISABLE_CONTEXT_SWITCHING #else // ifdef CPPA_DISABLE_CONTEXT_SWITCHING
......
...@@ -90,7 +90,7 @@ intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> f ...@@ -90,7 +90,7 @@ intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> f
event_based_actor::event_based_actor(actor_state st) : super(st, true) { } event_based_actor::event_based_actor(actor_state st) : super(st, true) { }
resume_result event_based_resume::resume(local_actor* self, util::fiber*) { resume_result event_based_resume::resume(untyped_actor* self, util::fiber*) {
CPPA_LOG_TRACE("id = " << self->id() << ", state = " << static_cast<int>(state())); CPPA_LOG_TRACE("id = " << self->id() << ", state = " << static_cast<int>(state()));
CPPA_REQUIRE( self->state() == actor_state::ready CPPA_REQUIRE( self->state() == actor_state::ready
|| self->state() == actor_state::pending); || self->state() == actor_state::pending);
......
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