Commit 6d43a00a authored by Dominik Charousset's avatar Dominik Charousset

implemented fibers for boost >= 1.54

parent d6195bec
...@@ -103,40 +103,6 @@ class context_switching_resume { ...@@ -103,40 +103,6 @@ class context_switching_resume {
}; };
/*
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>
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();
}
}
*/
template<class Actor> template<class Actor>
void await_ready(Actor* self) { void await_ready(Actor* self) {
while (!self->has_next_message()) { while (!self->has_next_message()) {
......
...@@ -63,6 +63,7 @@ blocking_untyped_actor::timed_sync_send_tuple(const util::duration& rtime, ...@@ -63,6 +63,7 @@ blocking_untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
} }
void blocking_untyped_actor::quit(std::uint32_t reason) { void blocking_untyped_actor::quit(std::uint32_t reason) {
planned_exit_reason(reason);
throw actor_exited(reason); throw actor_exited(reason);
} }
......
...@@ -34,25 +34,30 @@ ...@@ -34,25 +34,30 @@
#include <iostream> #include <iostream>
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
#include "cppa/self.hpp" #include "cppa/exception.hpp"
namespace cppa { namespace cppa {
namespace policy { namespace policy {
void context_switching_resume::trampoline(void* this_ptr) { void context_switching_resume::trampoline(void* this_ptr) {
auto _this = reinterpret_cast<blocking_untyped_actor*>(this_ptr); auto _this = reinterpret_cast<blocking_untyped_actor*>(this_ptr);
bool cleanup_called = false; auto shut_actor_down = [_this](std::uint32_t reason) {
try { _this->act(); } if (_this->planned_exit_reason() == exit_reason::not_exited) {
catch (actor_exited&) { _this->planned_exit_reason(reason);
// cleanup already called by scheduled_actor::quit }
cleanup_called = true; _this->on_exit();
_this->cleanup(_this->planned_exit_reason());
};
try {
_this->act();
shut_actor_down(exit_reason::normal);
}
catch (actor_exited& e) {
shut_actor_down(e.reason());
} }
catch (...) { catch (...) {
_this->cleanup(exit_reason::unhandled_exception); shut_actor_down(exit_reason::unhandled_exception);
cleanup_called = true;
} }
if (!cleanup_called) _this->cleanup(exit_reason::normal);
_this->on_exit();
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
detail::yield(detail::yield_state::done); detail::yield(detail::yield_state::done);
} }
......
...@@ -103,7 +103,7 @@ inline void fc_make(fc_member& storage, fc_allocator& alloc, vg_member& vgm) { ...@@ -103,7 +103,7 @@ inline void fc_make(fc_member& storage, fc_allocator& alloc, vg_member& vgm) {
reinterpret_cast<void*>( reinterpret_cast<void*>(
reinterpret_cast<intptr_t>(storage.fc_stack.base) - mss)); reinterpret_cast<intptr_t>(storage.fc_stack.base) - mss));
} }
#else #elif BOOST_VERSION < 105400
namespace ctx = boost::context; namespace ctx = boost::context;
typedef ctx::fcontext_t* fc_member; typedef ctx::fcontext_t* fc_member;
# if BOOST_VERSION < 105300 # if BOOST_VERSION < 105300
...@@ -122,6 +122,28 @@ inline void fc_make(fc_member& storage, fc_allocator& alloc, vg_member& vgm) { ...@@ -122,6 +122,28 @@ inline void fc_make(fc_member& storage, fc_allocator& alloc, vg_member& vgm) {
reinterpret_cast<void*>( reinterpret_cast<void*>(
reinterpret_cast<intptr_t>(storage->fc_stack.sp) - mss)); reinterpret_cast<intptr_t>(storage->fc_stack.sp) - mss));
} }
#else // BOOST_VERSION >= 105400
namespace ctx = boost::context;
typedef ctx::fcontext_t* fc_member;
typedef boost::coroutines::stack_context fc_make_result;
typedef boost::coroutines::stack_allocator fc_allocator;
inline void fc_jump(fc_member& from, fc_member& to, fiber_impl* ptr) {
ctx::jump_fcontext(from, to, (intptr_t) ptr);
}
inline fc_make_result fc_make(fc_member& storage, fc_allocator& alloc, vg_member& vgm) {
size_t mss = fc_allocator::minimum_stacksize();
fc_make_result sctx;
alloc.allocate(sctx, mss);
storage = ctx::make_fcontext(sctx.sp, sctx.size, fiber_trampoline);
vg_register(vgm,
storage->fc_stack.sp,
reinterpret_cast<void*>(
reinterpret_cast<intptr_t>(storage->fc_stack.sp) - mss));
return sctx;
}
inline void fc_destroy(fc_allocator& alloc, fc_make_result sctx) {
alloc.deallocate(sctx);
}
#endif #endif
} // namespace <anonymous> } // namespace <anonymous>
...@@ -161,11 +183,12 @@ struct converted_fiber : fiber_impl { ...@@ -161,11 +183,12 @@ struct converted_fiber : fiber_impl {
struct fun_fiber : fiber_impl { struct fun_fiber : fiber_impl {
fun_fiber(void (*fun)(void*), void* arg) : m_arg(arg), m_fun(fun) { fun_fiber(void (*fun)(void*), void* arg) : m_arg(arg), m_fun(fun) {
fc_make(m_ctx, m_alloc, m_vgm); m_make_res = fc_make(m_ctx, m_alloc, m_vgm);
} }
~fun_fiber() { ~fun_fiber() {
vg_deregister(m_vgm); vg_deregister(m_vgm);
fc_destroy(m_alloc, m_make_res);
} }
virtual void run() { virtual void run() {
...@@ -174,6 +197,7 @@ struct fun_fiber : fiber_impl { ...@@ -174,6 +197,7 @@ struct fun_fiber : fiber_impl {
void* m_arg; void* m_arg;
void (*m_fun)(void*); void (*m_fun)(void*);
fc_make_result m_make_res;
fc_allocator m_alloc; fc_allocator m_alloc;
vg_member m_vgm; vg_member m_vgm;
......
...@@ -734,7 +734,7 @@ void test_spawn() { ...@@ -734,7 +734,7 @@ void test_spawn() {
<< ", " << CPPA_TARG(buddy, to_string)); << ", " << CPPA_TARG(buddy, to_string));
self->become( self->become(
on_arg_match >> [=](int n, const string& s) { on_arg_match >> [=](int n, const string& s) {
self->send(buddy, n * 2, s); self->send(buddy, n * 2, s + " from " + name);
}, },
on(atom("done")) >> [=] { on(atom("done")) >> [=] {
self->quit(); self->quit();
...@@ -745,7 +745,7 @@ void test_spawn() { ...@@ -745,7 +745,7 @@ void test_spawn() {
auto bob = spawn(inflater, "Bob", joe); auto bob = spawn(inflater, "Bob", joe);
self->send(bob, 1, "hello actor"); self->send(bob, 1, "hello actor");
self->receive ( self->receive (
on(4, "hello actor") >> CPPA_CHECKPOINT_CB(), on(4, "hello actor from Bob from Joe") >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB() others() >> CPPA_UNEXPECTED_MSG_CB()
); );
// kill joe and bob // kill joe and bob
......
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