Commit 804047b1 authored by Dominik Charousset's avatar Dominik Charousset

bugfixes (passes test_spawn now)

parent f930ce71
......@@ -325,6 +325,12 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
virtual void act() = 0;
/**
* @brief Unwinds the stack by throwing an actor_exited exception.
* @throws actor_exited
*/
virtual void quit(std::uint32_t reason = exit_reason::normal);
private:
std::map<message_id, behavior> m_sync_handler;
......
......@@ -42,9 +42,9 @@ class channel;
class node_id;
class behavior;
class any_tuple;
class self_type;
class actor_addr;
class actor_proxy;
class scoped_actor;
class untyped_actor;
class abstract_actor;
class message_header;
......
......@@ -112,18 +112,6 @@ class functor_based_actor : public untyped_actor {
};
}
/*
template<class Actor, typename F, typename T0, typename... Ts>
void create(void*, F, T0&&, Ts&&...) {
// this overload acts as 'catch-all' rule to
// give the user a clear hint at what's wrong
// with the provided functor
static_assert(sizeof...(Ts) != sizeof...(Ts),
"spawn: provided functor must either return 'void'' "
"or 'cppa::behavior'");
}
*/
make_behavior_fun m_make_behavior;
};
......
......@@ -74,9 +74,8 @@ struct implicit_conversions {
typedef typename util::replace_type<
subtype3,
actor,
std::is_convertible<T, actor*>,
std::is_convertible<T, local_actor*>,
std::is_same<self_type, T>
std::is_convertible<T, abstract_actor*>,
std::is_same<scoped_actor, T>
>::type
type;
......
This diff is collapsed.
......@@ -31,6 +31,7 @@
#ifndef CPPA_LOGGING_HPP
#define CPPA_LOGGING_HPP
#include <cstring>
#include <sstream>
#include <iostream>
#include <execinfo.h>
......@@ -161,19 +162,25 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#define CPPA_LVL_NAME3() "DEBUG"
#define CPPA_LVL_NAME4() "TRACE"
#ifndef CPPA_LOG_LEVEL
# define CPPA_LOG_IMPL(lvlname, classname, funname, message) { \
#define CPPA_PRINT_ERROR_IMPL(lvlname, classname, funname, message) { \
std::cerr << "[" << lvlname << "] " << classname << "::" \
<< funname << ": " << message << "\nStack trace:\n"; \
void *array[10]; \
size_t size = backtrace(array, 10); \
backtrace_symbols_fd(array, size, 2); \
void* bt_array[20]; \
size_t size = backtrace(bt_array, 20); \
backtrace_symbols_fd(bt_array, size, 2); \
} CPPA_VOID_STMT
#ifndef CPPA_LOG_LEVEL
# define CPPA_LOG_IMPL(lvlname, classname, funname, message) \
CPPA_PRINT_ERROR_IMPL(lvlname, classname, funname, message)
# define CPPA_PUSH_AID(unused)
# define CPPA_SET_AID(unused) 0
# define CPPA_LOG_LEVEL 1
#else
# define CPPA_LOG_IMPL(lvlname, classname, funname, message) \
if (strcmp(lvlname, "ERROR") == 0) { \
CPPA_PRINT_ERROR_IMPL(lvlname, classname, funname, message); \
} \
::cppa::get_logger()->log(lvlname, classname, funname, __FILE__, \
__LINE__, (::cppa::oss_wr{} << message).str())
# define CPPA_PUSH_AID(aid_arg) \
......
......@@ -51,7 +51,6 @@ class message_header {
actor_addr sender;
channel receiver;
message_id id;
message_priority priority;
/**
* @brief An invalid message header without receiver or sender;
......@@ -64,14 +63,7 @@ class message_header {
*/
message_header(actor_addr source,
channel dest,
message_id mid = message_id::invalid,
message_priority prio = message_priority::normal);
/**
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = self</tt>.
*/
message_header(actor_addr source, channel dest, message_priority prio);
message_id mid = message_id::invalid);
void deliver(any_tuple msg) const;
......
......@@ -48,7 +48,7 @@ class message_id : util::comparable<message_id> {
static constexpr std::uint64_t response_flag_mask = 0x8000000000000000;
static constexpr std::uint64_t answered_flag_mask = 0x4000000000000000;
static constexpr std::uint64_t high_prioity_flag_mask = 0x1000000000000000;
static constexpr std::uint64_t high_prioity_flag_mask = 0x2000000000000000;
static constexpr std::uint64_t request_id_mask = 0x1FFFFFFFFFFFFFFF;
public:
......@@ -86,13 +86,23 @@ class message_id : util::comparable<message_id> {
}
inline message_id response_id() const {
return message_id(valid() ? m_value | response_flag_mask : 0);
// the response to a response is an asynchronous message
if (is_response()) return message_id{0};
return message_id{valid() ? m_value | response_flag_mask : 0};
}
inline message_id request_id() const {
return message_id(m_value & request_id_mask);
}
inline message_id with_high_priority() const {
return message_id(m_value | high_prioity_flag_mask);
}
inline message_id with_normal_priority() const {
return message_id(m_value & ~high_prioity_flag_mask);
}
inline void mark_as_answered() {
m_value |= answered_flag_mask;
}
......
......@@ -94,16 +94,19 @@ class event_based_resume {
auto ptr = d->next_message();
if (ptr) {
CPPA_REQUIRE(!d->bhvr_stack().empty());
auto bhvr = d->bhvr_stack().back();
auto mid = d->bhvr_stack().back_id();
if (d->invoke_message(ptr, bhvr, mid)) {
bool continue_from_cache = false;
if (d->invoke_message(ptr)) {
continue_from_cache = true;
if (actor_done() && done_cb()) {
CPPA_LOG_DEBUG("actor exited");
return resume_result::done;
}
while (d->invoke_message_from_cache()) {
// rinse and repeat
}
}
// add ptr to cache if invoke_message
// did not reset it
// did not reset it (i.e. skipped, but not dropped)
if (ptr) d->push_to_cache(std::move(ptr));
}
else {
......
......@@ -132,14 +132,15 @@ class invoke_policy {
}
enum filter_result {
normal_exit_signal,
non_normal_exit_signal,
expired_timeout_message,
expired_sync_response,
timeout_message,
timeout_response_message,
ordinary_message,
sync_response
normal_exit_signal, // an exit message with normal exit reason
non_normal_exit_signal, // an exit message with abnormal exit reason
expired_timeout_message, // an 'old & obsolete' timeout
inactive_timeout_message, // a currently inactive timeout
expired_sync_response, // a sync response that already timed out
timeout_message, // triggers currently active timeout
timeout_response_message, // triggers timeout of a sync message
ordinary_message, // an asynchronous message or sync. request
sync_response // a synchronous response
};
......@@ -169,7 +170,8 @@ class invoke_policy {
}
else if (v0 == atom("SYNC_TOUT")) {
CPPA_REQUIRE(!mid.valid());
return self->waits_for_timeout(v1) ? timeout_message
if (self->is_active_timeout(v1)) return timeout_message;
return self->waits_for_timeout(v1) ? inactive_timeout_message
: expired_timeout_message;
}
}
......@@ -181,7 +183,7 @@ class invoke_policy {
}
if (mid.is_response()) {
return (self->awaits(mid)) ? sync_response
: expired_sync_response;
: expired_sync_response;
}
return ordinary_message;
}
......@@ -308,6 +310,10 @@ class invoke_policy {
CPPA_LOG_DEBUG("dropped expired timeout message");
return hm_drop_msg;
}
case inactive_timeout_message: {
CPPA_LOG_DEBUG("skipped inactive timeout message");
return hm_skip_msg;
}
case non_normal_exit_signal: {
CPPA_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
......@@ -316,7 +322,8 @@ class invoke_policy {
}
case timeout_message: {
CPPA_LOG_DEBUG("handle timeout message");
self->handle_timeout(fun);
auto tid = node->msg.get_as<std::uint32_t>(1);
self->handle_timeout(fun, tid);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
......
......@@ -52,11 +52,9 @@ namespace cppa { namespace policy {
class nestable_invoke : public invoke_policy<nestable_invoke> {
typedef std::unique_lock<std::mutex> lock_type;
public:
static inline bool hm_should_skip(mailbox_element* node) {
inline bool hm_should_skip(mailbox_element* node) {
return node->marked;
}
......@@ -82,8 +80,6 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
self->pop_timeout();
}
typedef std::chrono::high_resolution_clock::time_point timeout_type;
};
} } // namespace cppa::policy
......
......@@ -4,6 +4,8 @@
#include <chrono>
#include <utility>
#include "cppa/exception.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/policy/resume_policy.hpp"
namespace cppa {
......@@ -28,7 +30,21 @@ class no_resume {
mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
inline resumable::resume_result resume(util::fiber*) {
this->act();
auto done_cb = [=](std::uint32_t reason) {
this->planned_exit_reason(reason);
this->on_exit();
this->cleanup(reason);
};
try {
this->act();
done_cb(exit_reason::normal);
}
catch (actor_exited& e) {
done_cb(e.reason());
}
catch (...) {
done_cb(exit_reason::unhandled_exception);
}
return resumable::done;
}
};
......
......@@ -106,20 +106,24 @@ class no_scheduling {
template<class Actor>
void launch(Actor* self) {
CPPA_PUSH_AID(self->id());
CPPA_LOG_TRACE(CPPA_ARG(self));
CPPA_REQUIRE(self != nullptr);
get_actor_registry()->inc_running();
intrusive_ptr<Actor> mself{self};
std::thread([=] {
CPPA_PUSH_AID(self->id());
CPPA_PUSH_AID(mself->id());
CPPA_LOG_TRACE("");
auto guard = util::make_scope_guard([] {
get_actor_registry()->dec_running();
});
util::fiber fself;
for (;;) {
await_data(self);
self->set_state(actor_state::ready);
if (self->resume(&fself) == resumable::done) {
mself->set_state(actor_state::ready);
if (mself->resume(&fself) == resumable::done) {
return;
}
// await new data before resuming actor
await_data(mself.get());
}
}).detach();
}
......
......@@ -31,6 +31,8 @@
#ifndef NOT_PRIORITIZING_HPP
#define NOT_PRIORITIZING_HPP
#include <iterator>
#include "cppa/mailbox_element.hpp"
#include "cppa/policy/priority_policy.hpp"
......@@ -72,6 +74,23 @@ class not_prioritizing {
m_cache.erase(iter);
}
inline bool cache_empty() const {
return m_cache.empty();
}
inline unique_mailbox_element_pointer cache_take_first() {
auto tmp = std::move(m_cache.front());
m_cache.erase(m_cache.begin());
return std::move(tmp);
}
template<typename Iterator>
inline void cache_prepend(Iterator first, Iterator last) {
auto mi = std::make_move_iterator(first);
auto me = std::make_move_iterator(last);
m_cache.insert(m_cache.begin(), mi, me);
}
private:
cache_type m_cache;
......
......@@ -49,16 +49,30 @@ class prioritizing {
template<class Actor>
unique_mailbox_element_pointer next_message(Actor* self) {
return unique_mailbox_element_pointer{self->mailbox().try_pop()};
if (!m_high.empty()) return take_first(m_high);
// read whole mailbox
unique_mailbox_element_pointer tmp{self->mailbox().try_pop()};
while (tmp) {
if (tmp->mid.is_high_priority()) m_high.push_back(std::move(tmp));
else m_low.push_back(std::move(tmp));
tmp.reset(self->mailbox().try_pop());
}
if (!m_high.empty()) return take_first(m_high);
if (!m_low.empty()) return take_first(m_low);
return unique_mailbox_element_pointer{};
}
template<class Actor>
inline bool has_next_message(Actor* self) {
return self->mailbox().can_fetch_more();
return !m_high.empty() || !m_low.empty() || self->mailbox().can_fetch_more();
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
m_cache.push_back(std::move(ptr));
if (ptr->mid.is_high_priority()) {
// insert before first element with low priority
m_cache.insert(cache_low_begin(), std::move(ptr));
}
else m_cache.push_back(std::move(ptr));
}
inline cache_iterator cache_begin() {
......@@ -73,47 +87,53 @@ class prioritizing {
m_cache.erase(iter);
}
private:
cache_type m_cache;
/*
mailbox_element* try_pop() {
auto result = m_high_priority_mailbox.try_pop();
return (result) ? result : this->m_mailbox.try_pop();
inline bool cache_empty() const {
return m_cache.empty();
}
template<typename... Ts>
prioritizing(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
protected:
typedef prioritizing combined_type;
inline unique_mailbox_element_pointer cache_take_first() {
return take_first(m_cache);
}
void cleanup(std::uint32_t reason) {
detail::sync_request_bouncer f{reason};
m_high_priority_mailbox.close(f);
Base::cleanup(reason);
template<typename Iterator>
inline void cache_prepend(Iterator first, Iterator last) {
using std::make_move_iterator;
// split input range between high and low priority messages
cache_type high;
cache_type low;
for (; first != last; ++first) {
if ((*first)->mid.is_high_priority()) high.push_back(std::move(*first));
else low.push_back(std::move(*first));
}
// prepend high priority messages
m_cache.insert(m_cache.begin(),
make_move_iterator(high.begin()),
make_move_iterator(high.end()));
// insert low priority messages after high priority messages
m_cache.insert(cache_low_begin(),
make_move_iterator(low.begin()),
make_move_iterator(low.end()));
}
bool mailbox_empty() {
return m_high_priority_mailbox.empty()
&& this->m_mailbox.empty();
private:
cache_iterator cache_low_begin() {
// insert before first element with low priority
return std::find_if(m_cache.begin(), m_cache.end(),
[](const unique_mailbox_element_pointer& e) {
return !e->mid.is_high_priority();
});
}
void enqueue(const message_header& hdr, any_tuple msg) {
typename Base::mailbox_type* mbox = nullptr;
if (hdr.priority == message_priority::high) {
mbox = &m_high_priority_mailbox;
}
else {
mbox = &this->m_mailbox;
}
this->enqueue_impl(*mbox, hdr, std::move(msg));
inline unique_mailbox_element_pointer take_first(cache_type& from) {
auto tmp = std::move(from.front());
from.erase(from.begin());
return std::move(tmp);
}
typename Base::mailbox_type m_high_priority_mailbox;
*/
cache_type m_cache;
cache_type m_high;
cache_type m_low;
};
......
......@@ -47,24 +47,24 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
public:
static inline bool hm_should_skip(mailbox_element*) {
inline bool hm_should_skip(mailbox_element*) {
return false;
}
template<class Actor>
static inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
auto previous = self->current_node();
self->current_node(node);
return previous;
}
template<class Actor>
static inline void hm_cleanup(Actor* self, mailbox_element*) {
inline void hm_cleanup(Actor* self, mailbox_element*) {
self->current_node(self->dummy_node());
}
template<class Actor>
static inline void hm_revert(Actor* self, mailbox_element* previous) {
inline void hm_revert(Actor* self, mailbox_element* previous) {
self->current_node(previous);
}
......
......@@ -49,7 +49,7 @@ template<class Derived, class Base = untyped_actor>
class sb_actor : public Base {
static_assert(std::is_base_of<untyped_actor, Base>::value,
"Base must be either untyped_actor or a derived type");
"Base must be untyped_actor or a derived type");
protected:
......@@ -58,7 +58,7 @@ class sb_actor : public Base {
public:
/**
* @brief Overrides {@link untyped_actor::init()} and sets
* @brief Overrides {@link untyped_actor::make_behavior()} and sets
* the initial actor behavior to <tt>Derived::init_state</tt>.
*/
behavior make_behavior() override {
......
......@@ -88,21 +88,6 @@ class scheduler {
virtual void enqueue(resumable*) = 0;
/**
* @brief Informs the scheduler about a converted context
* (a thread that acts as actor).
* @note Calls <tt>what->attach(...)</tt>.
*/
virtual void register_converted_context(abstract_actor* what);
/**
* @brief Informs the scheduler about a hidden (non-actor)
* context that should be counted by await_others_done().
* @returns An {@link attachable} that the hidden context has to destroy
* if his lifetime ends.
*/
virtual attachable* register_hidden_context();
template<typename Duration, typename... Data>
void delayed_send(message_header hdr,
const Duration& rel_time,
......
......@@ -42,6 +42,8 @@ class scoped_actor {
scoped_actor();
scoped_actor(const scoped_actor&) = delete;
explicit scoped_actor(bool hidden);
~scoped_actor();
......@@ -70,12 +72,16 @@ class scoped_actor {
return get()->address();
}
inline actor_addr address() const {
return get()->address();
}
private:
void init(bool hidden);
bool m_hidden;
actor_id m_prev;
actor_id m_prev; // used for logging/debugging purposes only
intrusive_ptr<blocking_untyped_actor> m_self;
};
......
......@@ -34,12 +34,14 @@
#include <type_traits>
#include "cppa/policy.hpp"
#include "cppa/logging.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/detail/proper_actor.hpp"
#include "cppa/detail/functor_based_actor.hpp"
#include "cppa/detail/implicit_conversions.hpp"
#include "cppa/detail/functor_based_blocking_actor.hpp"
namespace cppa {
......@@ -49,15 +51,8 @@ namespace cppa {
* @{
*/
/**
* @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link untyped_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options, typename... Ts>
actor spawn(Ts&&... args) {
actor spawn_impl(Ts&&... args) {
static_assert(std::is_base_of<untyped_actor, Impl>::value ||
(std::is_base_of<blocking_untyped_actor, Impl>::value &&
has_blocking_api_flag(Options)),
......@@ -68,6 +63,7 @@ actor spawn(Ts&&... args) {
"top-level spawns cannot have monitor or link flag");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag");
CPPA_LOGF_TRACE("spawn " << detail::demangle<Impl>());
/*
using scheduling_policy = typename std::conditional<
has_detach_flag(Options),
......@@ -86,7 +82,7 @@ actor spawn(Ts&&... args) {
typename std::conditional<
has_detach_flag(Options),
policy::no_resume,
policy::context_switching_resume
policy::no_resume //policy::context_switching_resume
>::type,
policy::event_based_resume
>::type;
......@@ -101,10 +97,42 @@ actor spawn(Ts&&... args) {
invoke_policy>;
using proper_impl = detail::proper_actor<Impl, policies>;
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
CPPA_PUSH_AID(ptr->id());
ptr->launch();
return ptr;
}
template<typename T>
struct spawn_fwd {
static inline T& fwd(T& arg) { return arg; }
static inline const T& fwd(const T& arg) { return arg; }
static inline T&& fwd(T&& arg) { return std::move(arg); }
};
template<typename T, typename... Ts>
struct spawn_fwd<T (Ts...)> {
typedef T (*pointer) (Ts...);
static inline pointer fwd(pointer arg) { return arg; }
};
template<>
struct spawn_fwd<scoped_actor> {
template<typename T>
static inline actor fwd(T& arg) { return arg; }
};
/**
* @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link untyped_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options, typename... Ts>
actor spawn(Ts&&... args) {
return spawn_impl<Impl, Options>(spawn_fwd<typename util::rm_const_and_ref<Ts>::type>::fwd(std::forward<Ts>(args))...);
}
/**
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
......
......@@ -31,6 +31,7 @@
#ifndef CPPA_DURATION_HPP
#define CPPA_DURATION_HPP
#include <string>
#include <chrono>
#include <cstdint>
......@@ -97,6 +98,8 @@ class duration {
*/
inline bool is_zero() const { return count == 0; }
std::string to_string() const;
time_unit unit;
std::uint32_t count;
......
......@@ -204,7 +204,7 @@ void abstract_actor::cleanup(std::uint32_t reason) {
auto msg = make_any_tuple(atom("EXIT"), reason);
CPPA_LOGM_DEBUG("cppa::actor", "send EXIT to " << mlinks.size() << " links");
for (auto& aptr : mlinks) {
aptr->enqueue({address(), aptr, message_priority::high}, msg);
aptr->enqueue({address(), aptr, message_id{}.with_high_priority()}, msg);
}
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
......
......@@ -28,6 +28,7 @@
\******************************************************************************/
#include "cppa/logging.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp"
#include "cppa/blocking_untyped_actor.hpp"
......@@ -61,4 +62,8 @@ blocking_untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
return {nri.response_id(), this};
}
void blocking_untyped_actor::quit(std::uint32_t reason) {
throw actor_exited(reason);
}
} // namespace cppa
......@@ -28,6 +28,8 @@
\******************************************************************************/
#include <sstream>
#include "cppa/util/duration.hpp"
namespace {
......@@ -45,4 +47,17 @@ bool operator==(const duration& lhs, const duration& rhs) {
: ui64_val(lhs) == ui64_val(rhs));
}
std::string duration::to_string() const {
if (unit == time_unit::none) return "-invalid-";
std::ostringstream oss;
oss << count;
switch (unit) {
default: oss << "???"; break;
case time_unit::seconds: oss << "s"; break;
case time_unit::milliseconds: oss << "ms"; break;
case time_unit::microseconds: oss << "us"; break;
}
return oss.str();
}
} } // namespace cppa::util
......@@ -28,6 +28,7 @@
\******************************************************************************/
#include "cppa/logging.hpp"
#include "cppa/detail/functor_based_blocking_actor.hpp"
namespace cppa {
......@@ -38,6 +39,7 @@ void functor_based_blocking_actor::create(blocking_untyped_actor*, act_fun fun)
}
void functor_based_blocking_actor::act() {
CPPA_LOG_TRACE("");
m_act(this);
}
......
......@@ -57,7 +57,7 @@ class down_observer : public attachable {
void actor_exited(std::uint32_t reason) {
if (m_observer) {
auto ptr = detail::actor_addr_cast<abstract_actor>(m_observer);
message_header hdr{m_observed, ptr, message_priority::high};
message_header hdr{m_observed, ptr, message_id{}.with_high_priority()};
hdr.deliver(make_any_tuple(atom("DOWN"), reason));
}
}
......@@ -135,14 +135,18 @@ void local_actor::reply_message(any_tuple&& what) {
void local_actor::forward_message(const actor& dest, message_priority p) {
if (!dest) return;
auto& id = m_current_node->mid;
detail::raw_access::get(dest)->enqueue({m_current_node->sender, detail::raw_access::get(dest), id, p}, m_current_node->msg);
auto id = (p == message_priority::high)
? m_current_node->mid.with_high_priority()
: m_current_node->mid.with_normal_priority();
detail::raw_access::get(dest)->enqueue({m_current_node->sender, detail::raw_access::get(dest), id}, m_current_node->msg);
// treat this message as asynchronous message from now on
id = message_id{};
m_current_node->mid = message_id{};
}
void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) {
dest.enqueue({address(), dest, prio}, std::move(what));
message_id id;
if (prio == message_priority::high) id = id.with_high_priority();
dest.enqueue({address(), dest, id}, std::move(what));
}
void local_actor::send_tuple(const channel& dest, any_tuple what) {
......@@ -167,6 +171,7 @@ response_promise local_actor::make_response_promise() {
}
void local_actor::cleanup(std::uint32_t reason) {
CPPA_LOG_TRACE(CPPA_ARG(reason));
m_subscriptions.clear();
super::cleanup(reason);
}
......
......@@ -35,20 +35,13 @@ namespace cppa {
message_header::message_header(actor_addr source,
channel dest,
message_id mid,
message_priority prio)
: sender(source), receiver(dest), id(mid), priority(prio) { }
message_header::message_header(actor_addr source,
channel dest,
message_priority prio)
: sender(source), receiver(dest), priority(prio) { }
message_id mid)
: sender(source), receiver(dest), id(mid) { }
bool operator==(const message_header& lhs, const message_header& rhs) {
return lhs.sender == rhs.sender
&& lhs.receiver == rhs.receiver
&& lhs.id == rhs.id
&& lhs.priority == rhs.priority;
&& lhs.id == rhs.id;
}
bool operator!=(const message_header& lhs, const message_header& rhs) {
......
......@@ -57,12 +57,6 @@ typedef std::chrono::high_resolution_clock hrc;
typedef hrc::time_point time_point;
struct exit_observer : cppa::attachable {
~exit_observer() { get_actor_registry()->dec_running(); }
void actor_exited(std::uint32_t) { }
bool matches(const token&) { return false; }
};
typedef policy::policies<policy::no_scheduling, policy::not_prioritizing,
policy::no_resume, policy::nestable_invoke>
timer_actor_policies;
......@@ -284,18 +278,6 @@ actor scheduler::delayed_send_helper() {
return m_helper->m_timer.get();
}
void scheduler::register_converted_context(abstract_actor* what) {
if (what) {
get_actor_registry()->inc_running();
what->address()->attach(attachable_ptr{new exit_observer});
}
}
attachable* scheduler::register_hidden_context() {
get_actor_registry()->inc_running();
return new exit_observer;
}
void set_scheduler(scheduler* sched) {
if (detail::singleton_manager::set_scheduler(sched) == false) {
throw std::runtime_error("scheduler already set");
......
......@@ -57,10 +57,10 @@ blocking_untyped_actor* alloc() {
void scoped_actor::init(bool hidden) {
m_hidden = hidden;
m_self.reset(alloc());
if (!hidden) {
m_prev = CPPA_SET_AID(m_self->id());
if (!m_hidden) {
get_actor_registry()->inc_running();
}
m_prev = CPPA_SET_AID(m_self->id());
}
......@@ -75,8 +75,8 @@ scoped_actor::scoped_actor(bool hidden) {
scoped_actor::~scoped_actor() {
if (!m_hidden) {
get_actor_registry()->dec_running();
CPPA_SET_AID(m_prev);
}
CPPA_SET_AID(m_prev);
}
} // namespace cppa
......@@ -83,6 +83,16 @@ class string_serializer : public serializer {
out << value;
}
// make sure char's are treated as int8_t number, not as character
inline void operator()(const char& value) {
out << static_cast<int>(value);
}
// make sure char's are treated as int8_t number, not as character
inline void operator()(const unsigned char& value) {
out << static_cast<unsigned int>(value);
}
void operator()(const string& str) {
if (!suppress_quotes) out << "\"";
for (char c : str) {
......@@ -344,6 +354,18 @@ class string_deserializer : public deserializer {
istringstream s(str);
s >> what;
}
void operator()(char& what) {
istringstream s(str);
int tmp;
s >> tmp;
what = static_cast<char>(tmp);
}
void operator()(unsigned char& what) {
istringstream s(str);
unsigned int tmp;
s >> tmp;
what = static_cast<unsigned char>(tmp);
}
void operator()(string& what) {
what = str;
}
......
......@@ -298,16 +298,12 @@ void serialize_impl(const message_header& hdr, serializer* sink) {
serialize_impl(hdr.sender, sink);
serialize_impl(hdr.receiver, sink);
sink->write_value(hdr.id.integer_value());
sink->write_value(static_cast<uint32_t>(hdr.priority));
}
void deserialize_impl(message_header& hdr, deserializer* source) {
deserialize_impl(hdr.sender, source);
deserialize_impl(hdr.receiver, source);
hdr.id = message_id::from_integer_value(source->read<std::uint64_t>());
auto prio = source->read<std::uint32_t>();
//TODO: check whether integer is actually a valid priority
hdr.priority = static_cast<message_priority>(prio);
}
void serialize_impl(const node_id_ptr& ptr, serializer* sink) {
......
......@@ -53,7 +53,7 @@ vector<string> split(const string& str, char delim, bool keep_empties) {
}
void verbose_terminate() {
try { if (uncaught_exception()) throw; }
try { throw; }
catch (exception& e) {
CPPA_PRINTERR("terminate called after throwing "
<< to_verbose_string(e));
......
......@@ -336,7 +336,7 @@ int main(int argc, char** argv) {
}
auto c = self->spawn<client, monitored>(serv);
self->receive (
on(atom("DOWN"), arg_match) >> [=](uint32_t rsn) {
on(atom("DOWN"), arg_match) >> [&](uint32_t rsn) {
CPPA_CHECK_EQUAL(self->last_sender(), c);
CPPA_CHECK_EQUAL(rsn, exit_reason::normal);
}
......
......@@ -55,22 +55,27 @@ class event_testee : public sb_actor<event_testee> {
};
// quits after 5 timeouts
actor spawn_event_testee2() {
actor spawn_event_testee2(actor parent) {
struct impl : untyped_actor {
actor parent;
impl(actor parent) : parent(parent) { }
behavior wait4timeout(int remaining) {
CPPA_LOG_TRACE(CPPA_ARG(remaining));
return (
return {
after(chrono::milliseconds(50)) >> [=] {
if (remaining == 1) quit();
if (remaining == 1) {
send(parent, atom("t2done"));
quit();
}
else become(wait4timeout(remaining - 1));
}
);
};
}
behavior make_behavior() override {
return wait4timeout(5);
}
};
return spawn<impl>();
return spawn<impl>(parent);
}
struct chopstick : public sb_actor<chopstick> {
......@@ -153,7 +158,11 @@ class testee_actor {
// self->receives one timeout and quits
void testee1(untyped_actor* self) {
self->become(after(chrono::milliseconds(10)) >> [=] { self->unbecome(); });
CPPA_LOGF_TRACE("");
self->become(after(chrono::milliseconds(10)) >> [=] {
CPPA_LOGF_TRACE("");
self->unbecome();
});
}
void testee2(untyped_actor* self, actor other) {
......@@ -173,10 +182,10 @@ void testee2(untyped_actor* self, actor other) {
}
template<class Testee>
string behavior_test(actor et) {
scoped_actor self;
string result;
string behavior_test(scoped_actor& self, actor et) {
string testee_name = detail::to_uniform_name(typeid(Testee));
CPPA_LOGF_TRACE(CPPA_TARG(et, to_string) << ", " << CPPA_ARG(testee_name));
string result;
self->send(et, 1);
self->send(et, 2);
self->send(et, 3);
......@@ -192,12 +201,12 @@ string behavior_test(actor et) {
result = str;
},
after(chrono::minutes(1)) >> [&]() {
//after(chrono::seconds(2)) >> [&]() {
CPPA_LOGF_ERROR(testee_name << " does not reply");
throw runtime_error(testee_name + " does not reply");
}
);
self->send_exit(et, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
return result;
}
......@@ -584,17 +593,19 @@ void test_spawn() {
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
spawn_event_testee2();
spawn_event_testee2(self);
self->receive(on(atom("t2done")) >> CPPA_CHECKPOINT_CB());
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto cstk = spawn<chopstick>();
self->send(cstk, atom("take"), self);
self->receive (
on(atom("taken")) >> [&]() {
on(atom("taken")) >> [&] {
self->send(cstk, atom("put"), self);
self->send(cstk, atom("break"));
}
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
......@@ -608,7 +619,7 @@ void test_spawn() {
{
int i = 0;
self->receive_for(i, 10) (
on(atom("failure")) >> [] { }
on(atom("failure")) >> CPPA_CHECKPOINT_CB()
);
CPPA_CHECKPOINT();
}
......@@ -622,7 +633,7 @@ void test_spawn() {
}
);
vector<int> expected{9, 8, 7, 6, 5, 4, 3, 2, 1, 0};
CPPA_CHECK_EQUAL(util::join(values, ","), util::join(values, ","));
CPPA_CHECK_EQUAL(util::join(values, ","), util::join(expected, ","));
}
// terminate st
self->send_exit(st, exit_reason::user_shutdown);
......@@ -677,7 +688,6 @@ void test_spawn() {
on_arg_match >> [&](const string& str) -> string {
CPPA_CHECK(self->last_sender() != nullptr);
CPPA_CHECK_EQUAL(str, "nothing");
self->quit();
return "goodbye!";
},
after(chrono::minutes(1)) >> [] {
......@@ -717,7 +727,11 @@ void test_spawn() {
after(chrono::milliseconds(5)) >> CPPA_UNEXPECTED_TOUT_CB()
);
auto inflater = [](untyped_actor* self, const string&, actor buddy) {
CPPA_CHECKPOINT();
auto inflater = [](untyped_actor* self, const string& name, actor buddy) {
CPPA_LOGF_TRACE(CPPA_ARG(self) << ", " << CPPA_ARG(name)
<< ", " << CPPA_TARG(buddy, to_string));
self->become(
on_arg_match >> [=](int n, const string& s) {
self->send(buddy, n * 2, s);
......@@ -736,8 +750,8 @@ void test_spawn() {
);
// kill joe and bob
auto poison_pill = make_any_tuple(atom("done"));
anon_send(joe, poison_pill);
anon_send(bob, poison_pill);
anon_send_tuple(joe, poison_pill);
anon_send_tuple(bob, poison_pill);
self->await_all_other_actors_done();
function<actor (const string&, const actor&)> spawn_next;
......@@ -789,10 +803,11 @@ void test_spawn() {
self->send_exit(a1, exit_reason::user_shutdown);
self->send_exit(a2, exit_reason::user_shutdown);
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto res1 = behavior_test<testee_actor>(spawn<blocking_api>(testee_actor{}));
auto res1 = behavior_test<testee_actor>(self, spawn<blocking_api>(testee_actor{}));
CPPA_CHECK_EQUAL("wait4int", res1);
CPPA_CHECK_EQUAL(behavior_test<event_testee>(spawn<event_testee>()), "wait4int");
CPPA_CHECK_EQUAL(behavior_test<event_testee>(self, spawn<event_testee>()), "wait4int");
// create some actors linked to one single actor
// and kill them all through killing the link
......
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