Commit 22d4342c authored by Dominik Charousset's avatar Dominik Charousset

maintenance

parent c0705e59
......@@ -21,7 +21,6 @@ set(LIBCPPA_SRC
src/binary_serializer.cpp
src/channel.cpp
src/thread_mapped_actor.cpp
src/cppa.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
......
......@@ -202,7 +202,6 @@ src/binary_deserializer.cpp
src/binary_serializer.cpp
src/channel.cpp
src/thread_mapped_actor.cpp
src/cppa.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
......
......@@ -535,78 +535,63 @@ channel_ptr& operator<<(channel_ptr& whom, const any_tuple& what);
#else
template<class C, typename Arg0, typename... Args>
void send(intrusive_ptr<C>& whom, const Arg0& arg0, const Args&... args) {
static_assert(std::is_base_of<channel, C>::value, "C is not a channel");
if (whom) self->send_message(whom.get(), make_cow_tuple(arg0, args...));
}
namespace detail {
template<class C, typename Arg0, typename... Args>
void send(intrusive_ptr<C>&& whom, const Arg0& arg0, const Args&... args) {
static_assert(std::is_base_of<channel, C>::value, "C is not a channel");
intrusive_ptr<C> tmp(std::move(whom));
send(tmp, arg0, args...);
inline void send_impl(channel* whom, any_tuple&& what) {
if (whom != nullptr) whom->enqueue(self.get(), std::move(what));
}
// matches "send(this, ...)" in event-based actors
template<typename Arg0, typename... Args>
inline void send(local_actor* whom, const Arg0& arg0, const Args&... args) {
CPPA_REQUIRE(whom != nullptr);
whom->enqueue(whom, make_cow_tuple(arg0, args...));
inline void send_tpl_impl(channel* whom, Arg0&& arg0, Args&&... args) {
if (whom != nullptr) {
whom->enqueue(self.get(), make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Args>(args)...));
}
}
inline void send_impl(actor* whom, any_tuple&& what) {
if (whom) self->send_message(whom, std::move(what));
}
// matches send(self, ...);
template<typename Arg0, typename... Args>
inline void send(const self_type&, const Arg0& arg0, const Args&... args) {
send(static_cast<local_actor*>(self), arg0, args...);
inline void send_tpl_impl(actor* whom, Arg0&& arg0, Args&&... args) {
if (whom) {
self->send_message(whom, make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Args>(args)...));
}
}
} // namespace detail
template<class C>
typename std::enable_if<
std::is_base_of<channel, C>::value,
intrusive_ptr<C>&
>::type
operator<<(intrusive_ptr<C>& whom, const any_tuple& what) {
if (whom) self->send_message(whom.get(), what);
typename std::enable_if<std::is_base_of<channel, C>::value,
const intrusive_ptr<C>& >::type
operator<<(const intrusive_ptr<C>& whom, any_tuple what) {
detail::send_impl(whom.get(), std::move(what));
return whom;
}
template<class C>
typename std::enable_if<
std::is_base_of<channel, C>::value,
intrusive_ptr<C>
>::type
operator<<(intrusive_ptr<C>&& whom, const any_tuple& what) {
intrusive_ptr<C> tmp(std::move(whom));
tmp << what;
return std::move(tmp);
inline const self_type& operator<<(const self_type& s, any_tuple what) {
detail::send_impl(s.get(), std::move(what));
return s;
}
template<class C>
typename std::enable_if<
std::is_base_of<channel, C>::value,
intrusive_ptr<C>&
>::type
operator<<(intrusive_ptr<C>& whom, any_tuple&& what) {
if (whom) self->send_message(whom.get(), std::move(what));
return whom;
}
template<class C>
typename std::enable_if<
std::is_base_of<channel, C>::value,
intrusive_ptr<C>
>::type
operator<<(intrusive_ptr<C>&& whom, any_tuple&& what) {
intrusive_ptr<C> tmp(std::move(whom));
tmp << std::move(what);
return std::move(tmp);
template<class C, typename Arg0, typename... Args>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send(const intrusive_ptr<C>& whom, Arg0&& arg0, Args&&... args) {
detail::send_tpl_impl(whom.get(),
std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
}
const self_type& operator<<(const self_type& s, const any_tuple& what);
const self_type& operator<<(const self_type& s, any_tuple&& what);
// matches "send(this, ...)" in event-based actors
template<typename Arg0, typename... Args>
inline void send(local_actor* whom, Arg0&& arg0, Args&&... args) {
detail::send_tpl_impl(whom,
std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
}
#endif // CPPA_DOCUMENTATION
......
......@@ -92,12 +92,23 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T> >,
std::swap(m_ptr, other.m_ptr);
}
/**
* @brief Returns the raw pointer without modifying reference count.
*/
pointer release() {
auto result = m_ptr;
m_ptr = nullptr;
return result;
}
/**
* @brief Sets this pointer to @p ptr without modifying reference count.
*/
void adopt(pointer ptr) {
reset();
m_ptr = ptr;
}
void reset(pointer new_value = nullptr) {
if (m_ptr && !m_ptr->deref()) delete m_ptr;
set_ptr(new_value);
......@@ -136,6 +147,7 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T> >,
inline pointer operator->() const { return m_ptr; }
inline reference operator*() const { return *m_ptr; }
inline bool operator!() const { return m_ptr == nullptr; }
inline explicit operator bool() const { return m_ptr != nullptr; }
inline ptrdiff_t compare(const_pointer ptr) const {
......
......@@ -44,7 +44,7 @@ namespace cppa {
*/
extern local_actor* self;
#else
#else // CPPA_DOCUMENTATION
class local_actor;
......@@ -56,6 +56,8 @@ class self_type : public convertible<self_type, actor*> {
public:
typedef local_actor* pointer;
constexpr self_type() { }
// "inherited" from convertible<...>
......@@ -63,38 +65,54 @@ class self_type : public convertible<self_type, actor*> {
return convert_impl();
}
inline pointer get() const {
return get_impl();
}
// allow "self" wherever an local_actor or actor pointer is expected
inline operator local_actor*() const {
inline operator pointer() const {
return get_impl();
}
inline local_actor* operator->() const {
inline pointer operator->() const {
return get_impl();
}
// @pre get_unchecked() == nullptr
inline void set(local_actor* ptr) const {
inline void set(pointer ptr) const {
set_impl(ptr);
}
// @returns The current value without converting the calling context
// to an actor on-the-fly.
inline local_actor* unchecked() const {
inline pointer unchecked() const {
return get_unchecked_impl();
}
static void cleanup_fun(local_actor*);
inline pointer release() const {
return release_impl();
}
inline void adopt(pointer ptr) const {
adopt_impl(ptr);
}
static void cleanup_fun(pointer);
private:
static void set_impl(local_actor*);
static void set_impl(pointer);
static local_actor* get_unchecked_impl();
static pointer get_unchecked_impl();
static local_actor* get_impl();
static pointer get_impl();
static actor* convert_impl();
static pointer release_impl();
static void adopt_impl(pointer);
};
/*
......@@ -103,7 +121,31 @@ class self_type : public convertible<self_type, actor*> {
*/
constexpr self_type self;
#endif
class scoped_self_setter {
scoped_self_setter(const scoped_self_setter&) = delete;
scoped_self_setter& operator=(const scoped_self_setter&) = delete;
public:
inline scoped_self_setter(local_actor* new_value) {
m_original_value = self.release();
self.adopt(new_value);
}
inline ~scoped_self_setter() {
// restore self
static_cast<void>(self.release());
self.adopt(m_original_value);
}
private:
local_actor* m_original_value;
};
#endif // CPPA_DOCUMENTATION
} // namespace cppa
......
......@@ -47,7 +47,12 @@ context_switching_actor::context_switching_actor(std::function<void()> fun)
}
void context_switching_actor::run() {
if (m_behavior) m_behavior();
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->exec();
}
else if (m_behavior) {
m_behavior();
}
}
void context_switching_actor::trampoline(void* this_ptr) {
......@@ -98,7 +103,7 @@ void context_switching_actor::dequeue(behavior& bhvr) {
}
else if (m_recv_policy.invoke_from_cache(this, bhvr) == false) {
if (bhvr.timeout().is_zero()) {
for (auto e = m_mailbox.try_pop(); e != 0; e = m_mailbox.try_pop()) {
for (auto e = m_mailbox.try_pop(); e != 0; e = m_mailbox.try_pop()){
CPPA_REQUIRE(e->marked == false);
if (m_recv_policy.invoke(this, e, bhvr)) return;
}
......@@ -106,14 +111,14 @@ void context_switching_actor::dequeue(behavior& bhvr) {
}
else {
request_timeout(bhvr.timeout());
while (m_recv_policy.invoke(this, receive_node(), bhvr) == false) { }
while (m_recv_policy.invoke(this, receive_node(), bhvr) == false) {}
}
}
}
resume_result context_switching_actor::resume(util::fiber* from) {
using namespace detail;
self.set(this);
scoped_self_setter sss{this};
for (;;) {
switch (call(&m_fiber, from)) {
case yield_state::done: {
......@@ -149,7 +154,7 @@ void context_switching_actor::unbecome() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back();
}
else {
else if (m_scheduler != nullptr) {
quit();
}
}
......@@ -161,9 +166,12 @@ void context_switching_actor::do_become(behavior* bhvr, bool ownership, bool dis
}
else {
m_bhvr_stack_ptr.reset(new detail::behavior_stack);
// scheduler == nullptr if and only if become() is called inside init()
if (m_scheduler != nullptr) {
m_bhvr_stack_ptr->exec();
quit();
}
}
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#include "cppa/cppa.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
const self_type& operator<<(const self_type& s, const any_tuple& what) {
local_actor* sptr = s;
sptr->enqueue(sptr, what);
return s;
}
const self_type& operator<<(const self_type& s, any_tuple&& what) {
local_actor* sptr = s;
sptr->enqueue(sptr, std::move(what));
return s;
}
} // namespace cppa
......@@ -49,12 +49,12 @@ void event_based_actor::dequeue(partial_function&) {
}
resume_result event_based_actor::resume(util::fiber*) {
scoped_self_setter sss{this};
auto done_cb = [&]() {
m_state.store(abstract_scheduled_actor::done);
m_bhvr_stack.clear();
on_exit();
};
self.set(this);
try {
detail::recursive_queue_node* e;
for (;;) {
......@@ -88,12 +88,8 @@ resume_result event_based_actor::resume(util::fiber*) {
}
}
}
catch (actor_exited& what) {
cleanup(what.reason());
}
catch (...) {
cleanup(exit_reason::unhandled_exception);
}
catch (actor_exited& what) { cleanup(what.reason()); }
catch (...) { cleanup(exit_reason::unhandled_exception); }
done_cb();
return resume_result::actor_done;
}
......
......@@ -42,8 +42,15 @@ void scheduled_actor::init() { }
scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) {
CPPA_REQUIRE(sched != nullptr);
// init is called by the spawning actor, manipulate self to
// point to this actor
scoped_self_setter sss{this};
// initialize this actor
try { init(); }
catch (...) { }
// make sure scheduler is not set until init() is done
std::atomic_thread_fence(std::memory_order_seq_cst);
m_scheduler = sched;
init();
return this;
}
......
......@@ -40,33 +40,35 @@
namespace {
boost::thread_specific_ptr<cppa::local_actor> t_this_context(&cppa::self_type::cleanup_fun);
boost::thread_specific_ptr<cppa::local_actor> t_this_actor(&cppa::self_type::cleanup_fun);
} // namespace <anonymous>
namespace cppa {
void self_type::cleanup_fun(cppa::local_actor* what) {
if (what) {
auto ptr = dynamic_cast<thread_mapped_actor*>(what);
if (ptr) {
// make sure "unspawned" actors quit properly
ptr->cleanup(cppa::exit_reason::normal);
}
if (what && !what->deref()) delete what;
if (what->deref() == false) delete what;
}
}
void self_type::set_impl(local_actor* ptr) {
void self_type::set_impl(self_type::pointer ptr) {
if (ptr) ptr->ref();
t_this_context.reset(ptr);
t_this_actor.reset(ptr);
}
local_actor* self_type::get_impl() {
auto result = t_this_context.get();
self_type::pointer self_type::get_impl() {
auto result = t_this_actor.get();
if (result == nullptr) {
result = new thread_mapped_actor;
result->ref();
get_scheduler()->register_converted_context(result);
t_this_context.reset(result);
t_this_actor.reset(result);
}
return result;
}
......@@ -75,8 +77,16 @@ actor* self_type::convert_impl() {
return get_impl();
}
local_actor* self_type::get_unchecked_impl() {
return t_this_context.get();
self_type::pointer self_type::get_unchecked_impl() {
return t_this_actor.get();
}
void self_type::adopt_impl(self_type::pointer ptr) {
t_this_actor.reset(ptr);
}
self_type::pointer self_type::release_impl() {
return t_this_actor.release();
}
} // namespace cppa
//#define CPPA_VERBOSE_CHECK
#include <stack>
#include <chrono>
#include <iostream>
......@@ -14,6 +16,7 @@
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/util/callable_trait.hpp"
using std::cerr;
using std::cout;
......@@ -21,7 +24,6 @@ using std::endl;
using namespace cppa;
#if (__GNUC__ >= 4) && (__GNUC_MINOR__ >= 7)
struct simple_mirror : fsm_actor<simple_mirror> {
......@@ -324,6 +326,64 @@ auto actor_prototype(const Args&... args) -> actor_template<decltype(mexpr_conca
return {mexpr_concat(args...)};
}
class actor_factory {
public:
virtual ~actor_factory() { }
virtual actor_ptr spawn() = 0;
};
template<typename InitFun, typename... Members>
class simple_event_based_actor_impl : public event_based_actor {
public:
simple_event_based_actor_impl(InitFun fun) : m_fun(std::move(fun)) { }
void init() { apply(); }
private:
InitFun m_fun;
std::tuple<Members...> m_members;
void apply(typename std::add_pointer<Members>::type... args) {
m_fun(args...);
}
template<typename... Args>
void apply(Args... args) {
apply(args..., &std::get<sizeof...(Args)>(m_members));
}
};
template<typename InitFun, typename... Members>
class actor_tpl : public actor_factory {
InitFun m_init_fun;
public:
actor_tpl(InitFun fun) : m_init_fun(std::move(fun)) { }
actor_ptr spawn() {
return cppa::spawn(new simple_event_based_actor_impl<InitFun, Members...>(m_init_fun));
}
};
template<typename InitFun, class TypeList>
struct actor_tpl_from_type_list;
template<typename InitFun, typename... Ts>
struct actor_tpl_from_type_list<InitFun, util::type_list<Ts...> > {
typedef actor_tpl<InitFun, Ts...> type;
};
class str_wrapper {
str_wrapper() = delete;
......@@ -382,10 +442,28 @@ void foobar(const str_wrapper& x, const std::string& y) {
);
}
template<typename Fun>
std::unique_ptr<actor_factory> foobaz(Fun fun) {
typedef typename util::get_callable_trait<Fun>::type ctrait;
typedef typename ctrait::arg_types arg_types;
static_assert(util::tl_forall<arg_types, std::is_pointer>::value,
"Functor takes non-pointer arguments");
typedef typename util::tl_map<arg_types, std::remove_pointer>::type mems;
typedef typename actor_tpl_from_type_list<Fun, mems>::type tpl_type;
return std::unique_ptr<actor_factory>{new tpl_type(fun)};
}
size_t test__spawn() {
using std::string;
CPPA_TEST(test__spawn);
/*
actor_tpl<int, float, std::string> atpl;
atpl([](int* i, float* f, std::string* s) {
cout << "SUCCESS: " << *i << ", " << *f << ", \"" << *s << "\"" << endl;
});
*/
/*
actor_ptr tst = actor_facade<some_integer>()
.reply_to().when().with()
......@@ -419,6 +497,13 @@ size_t test__spawn() {
).spawn();
*/
CPPA_IF_VERBOSE(cout << "test echo actor ... " << std::flush);
auto mecho = spawn(echo_actor);
send(mecho, "hello echo");
receive(on("hello echo") >> []() { });
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
auto mirror = spawn(new simple_mirror);
CPPA_IF_VERBOSE(cout << "test mirror ... " << std::flush);
......@@ -459,13 +544,6 @@ size_t test__spawn() {
receive(after(std::chrono::seconds(1)) >> []() { });
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "test echo actor ... " << std::flush);
auto mecho = spawn(echo_actor);
send(mecho, "hello echo");
receive(on("hello echo") >> []() { });
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "testee1 ... " << std::flush);
spawn(testee1);
await_all_others_done();
......@@ -488,6 +566,35 @@ size_t test__spawn() {
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
/*
CPPA_IF_VERBOSE(cout << "test factory ... " << std::flush);
auto factory = foobaz([&](int* i, float*, std::string*) {
self->become (
on(atom("get_int")) >> [i]() {
reply(*i);
},
on(atom("set_int"), arg_match) >> [i](int new_value) {
*i = new_value;
},
on(atom("done")) >> []() {
self->quit();
}
);
});
auto foobaz_actor = factory->spawn();
send(foobaz_actor, atom("set_int"), 42);
send(foobaz_actor, atom("get_int"));
send(foobaz_actor, atom("done"));
receive (
on_arg_match >> [&](int value) {
CPPA_CHECK_EQUAL(42, value);
}
);
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
*/
{
bool invoked = false;
str_wrapper actor_facade_builder{"x"};
......
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