Commit f110c6d9 authored by Joseph Noir's avatar Joseph Noir

Merge branch 'unstable' of github.com:Neverlord/libcppa into unstable

Conflicts:
	src/opencl/command_dispatcher.cpp
parents 2c6e4511 488ca971
......@@ -72,7 +72,7 @@ endif ()
# set build type (evaluate ENABLE_DEBUG flag)
if (ENABLE_DEBUG)
set(CMAKE_BUILD_TYPE Debug)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DCPPA_DEBUG")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DCPPA_DEBUG_MODE")
endif (ENABLE_DEBUG)
if (CPPA_LOG_LEVEL)
......@@ -102,7 +102,6 @@ set(LIBCPPA_SRC
src/context_switching_actor.cpp
src/continuable_reader.cpp
src/continuable_writer.cpp
src/decorated_names_map.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
src/default_peer.cpp
......@@ -154,6 +153,7 @@ set(LIBCPPA_SRC
src/to_uniform_name.cpp
src/unicast_network.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp
src/weak_ptr_anchor.cpp
src/yield_interface.cpp)
......
......@@ -25,7 +25,6 @@ cppa/detail/behavior_impl.hpp
cppa/detail/behavior_stack.hpp
cppa/detail/boxed.hpp
cppa/detail/container_tuple_view.hpp
cppa/detail/decorated_names_map.hpp
cppa/detail/decorated_tuple.hpp
cppa/detail/default_uniform_type_info_impl.hpp
cppa/detail/demangle.hpp
......@@ -122,6 +121,7 @@ cppa/option.hpp
cppa/partial_function.hpp
cppa/primitive_type.hpp
cppa/primitive_variant.hpp
cppa/prioritizing.hpp
cppa/process_information.hpp
cppa/qtsupport/actor_widget_mixin.hpp
cppa/receive.hpp
......@@ -131,6 +131,7 @@ cppa/sb_actor.hpp
cppa/scheduled_actor.hpp
cppa/scheduler.hpp
cppa/self.hpp
cppa/send.hpp
cppa/serializer.hpp
cppa/singletons.hpp
cppa/spawn_options.hpp
......@@ -227,7 +228,6 @@ src/channel.cpp
src/context_switching_actor.cpp
src/continuable_reader.cpp
src/continuable_writer.cpp
src/decorated_names_map.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
src/default_peer.cpp
......@@ -283,6 +283,7 @@ src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp
src/unicast_network.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp
src/weak_ptr_anchor.cpp
src/yield_interface.cpp
unit_testing/ping_pong.cpp
......@@ -306,4 +307,3 @@ unit_testing/test_sync_send.cpp
unit_testing/test_tuple.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
cppa/prioritizing.hpp
......@@ -74,13 +74,6 @@ class actor : public channel {
public:
/**
* @brief Enqueues @p msg to the actor's mailbox and returns true if
* this actor is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_enqueue(const message_header& hdr, any_tuple msg);
/**
* @brief Attaches @p ptr to this actor.
*
......@@ -192,14 +185,14 @@ class actor : public channel {
*/
inline bool exited() const;
private:
// cannot be changed after construction
const actor_id m_id;
// you're either a proxy or you're not
const bool m_is_proxy;
private:
// initially exit_reason::not_exited
std::atomic<std::uint32_t> m_exit_reason;
......
......@@ -64,6 +64,15 @@ class channel : public ref_counted {
*/
virtual void enqueue(const message_header& hdr, any_tuple msg) = 0;
/**
* @brief Enqueues @p msg to the list of received messages and returns
* true if this is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_enqueue(const message_header& hdr, any_tuple msg);
protected:
virtual ~channel();
......
......@@ -55,7 +55,7 @@
#include <cstdio>
#include <cstdlib>
#ifdef CPPA_DEBUG
#ifdef CPPA_DEBUG_MODE
#include <execinfo.h>
#define CPPA_REQUIRE__(stmt, file, line) \
......@@ -70,9 +70,9 @@
if ((stmt) == false) { \
CPPA_REQUIRE__(#stmt, __FILE__, __LINE__); \
}((void) 0)
#else // CPPA_DEBUG
#else // CPPA_DEBUG_MODE
#define CPPA_REQUIRE(unused) ((void) 0)
#endif // CPPA_DEBUG
#endif // CPPA_DEBUG_MODE
#define CPPA_CRITICAL__(error, file, line) { \
printf("%s:%u: critical error: '%s'\n", file, line, error); \
......
......@@ -38,7 +38,6 @@
#include "cppa/stacked.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/yield_interface.hpp"
......@@ -77,6 +76,10 @@ class context_switching_actor : public extend<scheduled_actor,context_switching_
mailbox_element* await_message(const timeout_type& abs_time);
inline mailbox_element* try_pop() {
return m_mailbox.try_pop();
}
private:
// required by util::fiber
......
This diff is collapsed.
......@@ -151,13 +151,16 @@ default_behavior_impl<MatchExpr, F>* new_default_behavior_impl(const MatchExpr&
template<typename F>
class continuation_decorator : public behavior_impl {
typedef behavior_impl super;
public:
typedef typename behavior_impl::pointer pointer;
template<typename Fun>
continuation_decorator(Fun&& fun, pointer decorated)
: m_fun(std::forward<Fun>(fun)), m_decorated(std::move(decorated)) {
: super(decorated->timeout()), m_fun(std::forward<Fun>(fun))
, m_decorated(std::move(decorated)) {
CPPA_REQUIRE(m_decorated != nullptr);
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_DECORATED_NAMES_MAP_HPP
#define CPPA_DECORATED_NAMES_MAP_HPP
#include <map>
#include <string>
#include "cppa/detail/singleton_mixin.hpp"
namespace cppa { namespace detail {
class decorated_names_map : public singleton_mixin<decorated_names_map> {
friend class singleton_mixin<decorated_names_map>;
public:
// returns either a decorated version of @p demangled_name or
// @p demangled_name itself
const std::string& decorate(const std::string& demangled_name) const;
private:
decorated_names_map();
std::map<std::string, std::string> m_map;
};
} } // namespace cppa::detail
#endif // DECORATED_NAMES_MAP_HPP
......@@ -107,7 +107,7 @@ class decorated_tuple : public abstract_tuple {
decorated_tuple(cow_pointer_type d, const vector_type& v)
: super(false)
, m_decorated(std::move(d)), m_mapping(v) {
# ifdef CPPA_DEBUG
# ifdef CPPA_DEBUG_MODE
const cow_pointer_type& ptr = m_decorated; // prevent detaching
# endif
CPPA_REQUIRE(ptr->size() >= sizeof...(Ts));
......@@ -117,7 +117,7 @@ class decorated_tuple : public abstract_tuple {
decorated_tuple(cow_pointer_type d, size_t offset)
: super(false), m_decorated(std::move(d)) {
# ifdef CPPA_DEBUG
# ifdef CPPA_DEBUG_MODE
const cow_pointer_type& ptr = m_decorated; // prevent detaching
# endif
CPPA_REQUIRE((ptr->size() - offset) >= sizeof...(Ts));
......
......@@ -104,7 +104,7 @@ class event_based_actor_factory {
template<typename... Ts>
actor_ptr spawn(Ts&&... args) {
auto ptr = memory::create<impl>(m_init, m_on_exit, std::forward<Ts>(args)...);
auto ptr = make_counted<impl>(m_init, m_on_exit, std::forward<Ts>(args)...);
return get_scheduler()->exec(no_spawn_options, ptr);
}
......
......@@ -153,7 +153,7 @@ class receive_policy {
else if (!invoke_from_cache(client, bhvr)) {
if (bhvr.timeout().is_zero()) {
pointer e = nullptr;
while ((e = client->m_mailbox.try_pop()) != nullptr) {
while ((e = client->try_pop()) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) {
return; // done
......@@ -356,15 +356,15 @@ class receive_policy {
CPPA_CRITICAL("illegal filter result");
}
case normal_exit_signal: {
CPPA_LOG_DEBUG("dropped message: normal exit signal");
CPPA_LOGMF(CPPA_DEBUG, client, "dropped normal exit signal");
return hm_drop_msg;
}
case expired_sync_response: {
CPPA_LOG_DEBUG("dropped message: expired sync response");
CPPA_LOGMF(CPPA_DEBUG, client, "dropped expired sync response");
return hm_drop_msg;
}
case expired_timeout_message: {
CPPA_LOG_DEBUG("dropped message: expired timeout message");
CPPA_LOGMF(CPPA_DEBUG, client, "dropped expired timeout message");
return hm_drop_msg;
}
case non_normal_exit_signal: {
......@@ -388,7 +388,7 @@ class receive_policy {
if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = hm_begin(client, node, policy);
if (!fun(node->msg) && handle_sync_failure_on_mismatch) {
CPPA_LOG_WARNING("sync failure occured");
CPPA_LOGMF(CPPA_WARNING, client, "sync failure occured");
client->handle_sync_failure();
}
client->mark_arrived(awaited_response);
......@@ -407,9 +407,10 @@ class receive_policy {
auto id = node->mid;
auto sender = node->sender;
if (id.valid() && !id.is_answered() && sender) {
CPPA_LOG_WARNING("actor did not reply to a "
CPPA_LOGMF(CPPA_WARNING, client,
"actor did not reply to a "
"synchronous request message");
sender->enqueue({client, id.response_id()},
sender->enqueue({client, sender, id.response_id()},
make_any_tuple(atom("VOID")));
}
hm_cleanup(client, previous_node, policy);
......
......@@ -50,7 +50,6 @@ class empty_tuple;
class group_manager;
class abstract_tuple;
class actor_registry;
class decorated_names_map;
class uniform_type_info_map;
class singleton_manager {
......@@ -79,8 +78,6 @@ class singleton_manager {
static empty_tuple* get_empty_tuple();
static decorated_names_map* get_decorated_names_map();
static opencl::command_dispatcher* get_command_dispatcher();
private:
......
......@@ -48,13 +48,12 @@ struct sync_request_bouncer {
inline void operator()(const actor_ptr& sender, const message_id& mid) const {
CPPA_REQUIRE(rsn != exit_reason::not_exited);
if (mid.is_request() && sender != nullptr) {
actor_ptr nobody;
sender->enqueue({nobody, mid.response_id()},
sender->enqueue({nullptr, sender, mid.response_id()},
make_any_tuple(atom("EXITED"), rsn));
}
}
inline void operator()(const mailbox_element& e) const {
(*this)(e.sender.get(), e.mid);
(*this)(e.sender, e.mid);
}
};
......
......@@ -329,7 +329,7 @@ struct tdata<Head, Tail...> : tdata<Tail...> {
}
inline void* mutable_at(size_t p) {
# ifdef CPPA_DEBUG
# ifdef CPPA_DEBUG_MODE
if (p == 0) {
if (std::is_same<decltype(ptr_to(head)), const void*>::value) {
throw std::logic_error{"mutable_at with const head"};
......
......@@ -61,9 +61,9 @@ class thread_pool_scheduler : public scheduler {
void enqueue(scheduled_actor* what);
actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr);
virtual local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) override;
actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f);
virtual local_actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f) override;
private:
......
......@@ -32,57 +32,93 @@
#define CPPA_UNIFORM_TYPE_INFO_MAP_HPP
#include <set>
#include <map>
#include <string>
#include <utility> // std::pair
#include <utility>
#include <type_traits>
#include "cppa/cppa_fwd.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/detail/singleton_mixin.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
namespace cppa { class uniform_type_info; }
namespace cppa { namespace detail {
using mapped_type_list = util::type_list<
bool,
any_tuple,
atom_value,
actor_ptr,
channel_ptr,
group_ptr,
process_information_ptr,
message_header,
std::nullptr_t,
util::duration,
util::void_type,
double,
float,
long double,
std::string,
std::u16string,
std::u32string,
std::map<std::string,std::string>
>;
using zipped_type_list = util::tl_zip_with_index<mapped_type_list>::type;
// lookup table for built-in types
extern const char* mapped_type_names[][2];
template<typename T>
constexpr const char* mapped_name() {
return mapped_type_names[util::tl_index_of<zipped_type_list,T>::value][1];
}
const char* mapped_name_by_decorated_name(const char* decorated_type_name);
// lookup table for integer types
extern const char* mapped_int_names[][2];
template<typename T>
constexpr const char* mapped_int_name() {
return mapped_int_names[sizeof(T)][std::is_signed<T>::value ? 1 : 0];
}
class uniform_type_info_map_helper;
// note: this class is implemented in uniform_type_info.cpp
class uniform_type_info_map : public singleton_mixin<uniform_type_info_map> {
class uniform_type_info_map {
friend class uniform_type_info_map_helper;
friend class singleton_mixin<uniform_type_info_map>;
public:
typedef std::set<std::string> set_type;
typedef std::map<std::string, uniform_type_info*> uti_map_type;
typedef std::map<int, std::pair<set_type, set_type> > int_map_type;
typedef const uniform_type_info* pointer;
inline const int_map_type& int_names() const {
return m_ints;
}
virtual ~uniform_type_info_map();
const uniform_type_info* by_raw_name(const std::string& name) const;
virtual pointer by_uniform_name(const std::string& name) const = 0;
const uniform_type_info* by_uniform_name(const std::string& name) const;
virtual pointer by_rtti(const std::type_info& ti) const = 0;
std::vector<const uniform_type_info*> get_all() const;
virtual std::vector<pointer> get_all() const = 0;
// NOT thread safe!
bool insert(const std::set<std::string>& raw_names, uniform_type_info* uti);
private:
// maps raw typeid names to uniform type informations
uti_map_type m_by_rname;
virtual bool insert(uniform_type_info* uti) = 0;
// maps uniform names to uniform type informations
uti_map_type m_by_uname;
static uniform_type_info_map* create_singleton();
// maps sizeof(-integer_type-) to { signed-names-set, unsigned-names-set }
int_map_type m_ints;
inline void dispose() { delete this; }
uniform_type_info_map();
inline void destroy() { delete this; }
~uniform_type_info_map();
virtual void initialize() = 0;
};
......
......@@ -38,14 +38,14 @@ namespace cppa {
namespace detail {
template<class B, class D, CPPA_MIXIN... Ms>
template<class D, class B, CPPA_MIXIN... Ms>
struct extend_helper;
template<class B, class D>
struct extend_helper<B,D> { typedef D type; };
template<class D, class B>
struct extend_helper<D,B> { typedef B type; };
template<class B, class D, CPPA_MIXIN M, CPPA_MIXIN... Ms>
struct extend_helper<B,D,M,Ms...> : extend_helper<B,M<D,B>,Ms...> { };
template<class D, class B, CPPA_MIXIN M, CPPA_MIXIN... Ms>
struct extend_helper<D,B,M,Ms...> : extend_helper<D,M<B,D>,Ms...> { };
} // namespace detail
......@@ -57,7 +57,7 @@ struct extend_helper<B,D,M,Ms...> : extend_helper<B,M<D,B>,Ms...> { };
* Mixins in libcppa always have two template parameters: base type and
* derived type. This allows mixins to make use of the curiously recurring
* template pattern (CRTP). However, if none of the used mixins use CRTP,
* the second template argument can ignored (it is then set to Base).
* the second template argument can be ignored (it is then set to Base).
*/
template<class Base, class Derived = Base>
struct extend {
......
......@@ -36,6 +36,7 @@
#include <stdexcept>
#include <type_traits>
#include "cppa/memory_cached.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
......@@ -208,7 +209,14 @@ inline bool operator!=(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
* of {@link ref_counted} and wraps it in an {@link intrusive_ptr}.
*/
template<typename T, typename... Ts>
intrusive_ptr<T> make_counted(Ts&&... args) {
typename std::enable_if<is_memory_cached<T>::value,intrusive_ptr<T>>::type
make_counted(Ts&&... args) {
return {detail::memory::create<T>(std::forward<Ts>(args)...)};
}
template<typename T, typename... Ts>
typename std::enable_if<not is_memory_cached<T>::value,intrusive_ptr<T>>::type
make_counted(Ts&&... args) {
return {new T(std::forward<Ts>(args)...)};
}
......
......@@ -46,6 +46,7 @@
#include "cppa/message_header.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/message_priority.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/util/duration.hpp"
......@@ -55,6 +56,7 @@
namespace cppa {
// forward declarations
class self_type;
class scheduler;
class message_future;
class local_scheduler;
......@@ -88,10 +90,14 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
*/
class local_actor : public extend<actor>::with<memory_cached> {
friend class self_type;
typedef combined_type super;
public:
~local_actor();
/**
* @brief Causes this actor to subscribe to the group @p what.
*
......@@ -239,6 +245,12 @@ class local_actor : public extend<actor>::with<memory_cached> {
/** @cond PRIVATE */
inline message_id new_request_id() {
auto result = ++m_last_request_id;
m_pending_responses.push_back(result.response_id());
return result;
}
inline void handle_sync_timeout() {
if (m_sync_timeout_handler) m_sync_timeout_handler();
else quit(exit_reason::unhandled_sync_timeout);
......@@ -257,13 +269,7 @@ class local_actor : public extend<actor>::with<memory_cached> {
inline void dequeue_response(behavior&&, message_id);
template<bool Discard, typename... Ts>
void become(behavior_policy<Discard>, Ts&&... args);
template<typename T, typename... Ts>
void become(T arg, Ts&&... args);
inline void unbecome();
inline void do_unbecome();
local_actor(bool is_scheduled = false);
......@@ -271,14 +277,8 @@ class local_actor : public extend<actor>::with<memory_cached> {
inline bool chaining_enabled();
inline void send_message(channel* whom, any_tuple&& what);
inline void send_message(actor* whom, any_tuple&& what);
inline message_id send_sync_message(const actor_ptr& whom,
any_tuple&& what);
message_id send_timed_sync_message(const actor_ptr& whom,
message_id send_timed_sync_message(message_priority mp,
const actor_ptr& whom,
const util::duration& rel_time,
any_tuple&& what);
......@@ -302,16 +302,23 @@ class local_actor : public extend<actor>::with<memory_cached> {
inline detail::behavior_stack& bhvr_stack();
protected:
virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
inline void do_become(const behavior& bhvr, bool discard_old);
const char* debug_name() const;
void debug_name(std::string str);
protected:
inline void remove_handler(message_id id);
void cleanup(std::uint32_t reason);
// used *only* when compiled in debug mode
union { std::string m_debug_name; };
// true if this actor uses the chained_send optimization
bool m_chaining;
......@@ -397,17 +404,7 @@ inline actor_ptr& local_actor::last_sender() {
return m_current_node->sender;
}
template<bool Discard, typename... Ts>
inline void local_actor::become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
template<typename T, typename... Ts>
inline void local_actor::become(T arg, Ts&&... args) {
do_become(match_expr_convert(arg, std::forward<Ts>(args)...), true);
}
inline void local_actor::unbecome() {
inline void local_actor::do_unbecome() {
m_bhvr_stack.pop_async_back();
}
......@@ -415,33 +412,6 @@ inline bool local_actor::chaining_enabled() {
return m_chaining && !m_chained_actor;
}
inline void local_actor::send_message(channel* whom, any_tuple&& what) {
whom->enqueue(this, std::move(what));
}
inline void local_actor::send_message(actor* whom, any_tuple&& what) {
if (chaining_enabled()) {
if (whom->chained_enqueue(this, std::move(what))) {
m_chained_actor.reset(whom);
}
}
else whom->enqueue(this, std::move(what));
}
inline message_id local_actor::send_sync_message(const actor_ptr& whom, any_tuple&& what) {
auto id = ++m_last_request_id;
CPPA_REQUIRE(id.is_request());
if (chaining_enabled()) {
if (whom->chained_enqueue({this, id}, std::move(what))) {
chained_actor(whom);
}
}
else whom->enqueue({this, id}, std::move(what));
auto awaited_response = id.response_id();
m_pending_responses.push_back(awaited_response);
return awaited_response;
}
inline message_id local_actor::get_response_id() {
auto id = m_current_node->mid;
return (id.is_request()) ? id.response_id() : message_id();
......
This diff is collapsed.
......@@ -39,9 +39,6 @@
namespace cppa {
template<typename T>
struct has_blocking_receive;
template<class Base, class Subtype>
class mailbox_based : public Base {
......@@ -65,7 +62,7 @@ class mailbox_based : public Base {
template<typename... Ts>
mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
void cleanup(std::uint32_t reason) {
virtual void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
......
......@@ -31,6 +31,9 @@
#ifndef CPPA_MEMORY_CACHED_HPP
#define CPPA_MEMORY_CACHED_HPP
#include <utility>
#include <type_traits>
#include "cppa/detail/memory.hpp"
namespace cppa {
......@@ -47,6 +50,10 @@ class memory_cached : public Base {
template<typename>
friend class detail::basic_memory_cache;
public:
static constexpr bool is_memory_cached_type = true;
protected:
typedef memory_cached combined_type;
......@@ -74,6 +81,16 @@ class memory_cached : public Base {
};
template<typename T>
struct is_memory_cached {
template<class U, bool = U::is_memory_cached_type>
static std::true_type check(int);
template<class>
static std::false_type check(...);
public:
static constexpr bool value = decltype(check<T>(0))::value;
};
} // namespace cppa
#endif // CPPA_MEMORY_CACHED_HPP
......@@ -68,7 +68,7 @@ class message_future {
behavior cpy = ref;
ref = cpy.add_continuation(std::move(fun));
}
else CPPA_LOG_WARNING(".continue_with: failed to add continuation");
else CPPA_LOG_ERROR(".continue_with: failed to add continuation");
}
private:
......
......@@ -31,6 +31,7 @@
#ifndef CPPA_MESSAGE_HEADER_HPP
#define CPPA_MESSAGE_HEADER_HPP
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/message_id.hpp"
#include "cppa/message_priority.hpp"
......@@ -48,27 +49,69 @@ class message_header {
public:
actor_ptr sender;
actor_ptr receiver;
channel_ptr receiver;
message_id id;
message_priority priority;
/**
* @brief An invalid message header without receiver or sender;
**/
message_header() = default;
message_header(actor* sender);
message_header(const self_type& sender);
message_header(const actor_ptr& sender,
message_priority priority = message_priority::normal);
/**
* @brief Creates a message header with <tt>receiver = dest</tt>
* and <tt>sender = self</tt>.
**/
template<typename T>
message_header(intrusive_ptr<T> dest)
: sender(self), receiver(dest), priority(message_priority::normal) {
static_assert(std::is_convertible<T*,channel*>::value,
"illegal receiver");
}
template<typename T>
message_header(T* dest)
: sender(self), receiver(dest), priority(message_priority::normal) {
static_assert(std::is_convertible<T*,channel*>::value,
"illegal receiver");
}
message_header(const std::nullptr_t&);
/**
* @brief Creates a message header with <tt>receiver = self</tt>
* and <tt>sender = self</tt>.
**/
message_header(const self_type&);
/**
* @brief Creates a message header with <tt>receiver = dest</tt>
* and <tt>sender = self</tt>.
*/
message_header(channel_ptr dest,
message_id mid,
message_priority prio = message_priority::normal);
message_header(const actor_ptr &sender,
message_id id,
message_priority priority = message_priority::normal);
/**
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = self</tt>.
*/
message_header(channel_ptr dest, message_priority prio);
message_header(const actor_ptr& sender,
const actor_ptr& receiver,
message_id id = message_id::invalid,
message_priority priority = message_priority::normal);
/**
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = source</tt>.
*/
message_header(actor_ptr source,
channel_ptr 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_ptr source, channel_ptr dest, message_priority prio);
void deliver(any_tuple msg) const;
......
......@@ -107,7 +107,7 @@ class middleman_event_handler_base {
auto wptr = ptr->as_io();
if (wptr) fd = wptr->write_handle();
else {
CPPA_LOG_ERROR("ptr->downcast() returned nullptr");
CPPA_LOGMF(CPPA_ERROR, self, "ptr->downcast() returned nullptr");
return;
}
break;
......@@ -118,7 +118,7 @@ class middleman_event_handler_base {
if (wptr) {
auto wrfd = wptr->write_handle();
if (fd != wrfd) {
CPPA_LOG_DEBUG("read_handle != write_handle, split "
CPPA_LOGMF(CPPA_DEBUG, self, "read_handle != write_handle, split "
"into two function calls");
// split into two function calls
e = event::read;
......@@ -126,13 +126,13 @@ class middleman_event_handler_base {
}
}
else {
CPPA_LOG_ERROR("ptr->downcast() returned nullptr");
CPPA_LOGMF(CPPA_ERROR, self, "ptr->downcast() returned nullptr");
return;
}
break;
}
default:
CPPA_LOG_ERROR("invalid bitmask");
CPPA_LOGMF(CPPA_ERROR, self, "invalid bitmask");
return;
}
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
......@@ -163,7 +163,7 @@ class middleman_event_handler_base {
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
CPPA_LOG_DEBUG("new bitmask for "
CPPA_LOGMF(CPPA_DEBUG, self, "new bitmask for "
<< elem.ptr.get() << ": " << eb2str(mask));
if (iter == last || iter->fd != elem.fd) {
CPPA_LOG_INFO_IF(mask == event::none,
......
......@@ -82,7 +82,7 @@ class actor_facade<Ret(Args...)> : public actor {
result_mapping map_result) {
if (global_dims.empty()) {
auto str = "OpenCL kernel needs at least 1 global dimension.";
CPPA_LOGM_ERROR(detail::demangle(typeid(actor_facade)), str);
CPPA_LOGM_ERROR(detail::demangle(typeid(actor_facade)).c_str(), str);
throw std::runtime_error(str);
}
auto check_vec = [&](const dim_vec& vec, const char* name) {
......@@ -90,7 +90,7 @@ class actor_facade<Ret(Args...)> : public actor {
std::ostringstream oss;
oss << name << " vector is not empty, but "
<< "its size differs from global dimensions vector's size";
CPPA_LOGM_ERROR(detail::demangle<actor_facade>(), oss.str());
CPPA_LOGM_ERROR(detail::demangle<actor_facade>().c_str(), oss.str());
throw std::runtime_error(oss.str());
}
};
......@@ -104,7 +104,7 @@ class actor_facade<Ret(Args...)> : public actor {
if (err != CL_SUCCESS) {
std::ostringstream oss;
oss << "clCreateKernel: " << get_opencl_error(err);
CPPA_LOGM_ERROR(detail::demangle<actor_facade>(), oss.str());
CPPA_LOGM_ERROR(detail::demangle<actor_facade>().c_str(), oss.str());
throw std::runtime_error(oss.str());
}
return new actor_facade<Ret (Args...)>{dispatcher,
......@@ -169,7 +169,7 @@ class actor_facade<Ret(Args...)> : public actor {
m_local_dimensions,
m_map_result));
}
else { CPPA_LOG_ERROR("actor_facade::enqueue() tuple_cast failed."); }
else { CPPA_LOGMF(CPPA_ERROR, this, "actor_facade::enqueue() tuple_cast failed."); }
}
typedef std::vector<mem_ptr> args_vec;
......@@ -217,7 +217,7 @@ class actor_facade<Ret(Args...)> : public actor {
arg0.data(),
&err);
if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clCreateBuffer: " << get_opencl_error(err));
CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
}
else {
mem_ptr tmp;
......@@ -241,7 +241,7 @@ class actor_facade<Ret(Args...)> : public actor {
nullptr,
&err);
if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clCreateBuffer: " << get_opencl_error(err));
CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
}
else {
mem_ptr tmp;
......
......@@ -31,4 +31,57 @@
#ifndef PRIORITIZING_HPP
#define PRIORITIZING_HPP
#include <iostream>
#include "cppa/mailbox_element.hpp"
#include "cppa/message_priority.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
namespace cppa {
template<class Base, class Subtype>
class prioritizing : public Base {
public:
virtual mailbox_element* try_pop() override {
auto result = m_high_priority_mailbox.try_pop();
return (result) ? result : this->m_mailbox.try_pop();
}
template<typename... Ts>
prioritizing(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
protected:
typedef prioritizing combined_type;
virtual void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_high_priority_mailbox.close(f);
Base::cleanup(reason);
}
virtual bool mailbox_empty() override {
return m_high_priority_mailbox.empty()
&& this->m_mailbox.empty();
}
virtual void enqueue(const message_header& hdr, any_tuple msg) override {
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));
}
typename Base::mailbox_type m_high_priority_mailbox;
};
} // namespace cppa
#endif // PRIORITIZING_HPP
......@@ -120,11 +120,23 @@ inline bool equal(const process_information::node_id_type& node_id,
return equal(hash, node_id);
}
/**
* @brief A smart pointer type that manages instances of
* {@link process_information}.
* @relates process_information
*/
typedef intrusive_ptr<process_information> process_information_ptr;
/**
* @relates process_information
*/
std::string to_string(const process_information& what);
/**
* @relates process_information
*/
std::string to_string(const process_information_ptr& what);
/**
* @brief Converts a {@link process_information::node_id_type node_id}
* to a hexadecimal string.
......@@ -134,13 +146,6 @@ std::string to_string(const process_information& what);
*/
std::string to_string(const process_information::node_id_type& node_id);
/**
* @brief A smart pointer type that manages instances of
* {@link process_information}.
* @relates process_information
*/
typedef intrusive_ptr<process_information> process_information_ptr;
} // namespace cppa
#endif // CPPA_PROCESS_INFORMATION_HPP
......@@ -34,6 +34,7 @@
#include <utility>
#include <type_traits>
#include "cppa/util/dptr.hpp"
#include "cppa/event_based_actor.hpp"
namespace cppa {
......@@ -60,8 +61,8 @@ class sb_actor : public Base {
* @brief Overrides {@link event_based_actor::init()} and sets
* the initial actor behavior to <tt>Derived::init_state</tt>.
*/
virtual void init() {
this->become(static_cast<Derived*>(this)->init_state);
virtual void init() override {
become(util::dptr<Derived>(this)->init_state);
}
protected:
......
......@@ -64,9 +64,6 @@ enum scheduled_actor_type {
class scheduled_actor;
template<>
struct has_blocking_receive<scheduled_actor> : std::true_type { };
/**
* @brief A base class for cooperatively scheduled actors.
* @extends local_actor
......
......@@ -138,7 +138,8 @@ class scheduler {
util::duration{rel_time},
to,
std::move(data));
delayed_send_helper()->enqueue(self, std::move(tup));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
}
template<typename Duration, typename... Data>
......@@ -153,7 +154,8 @@ class scheduler {
to,
id,
std::move(data));
delayed_send_helper()->enqueue(self, std::move(tup));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
}
else {
this->delayed_send(to, rel_time, std::move(data));
......@@ -163,18 +165,18 @@ class scheduler {
/**
* @brief Executes @p ptr in this scheduler.
*/
virtual actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) = 0;
virtual local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) = 0;
/**
* @brief Creates a new actor from @p actor_behavior and executes it
* in this scheduler.
*/
virtual actor_ptr exec(spawn_options opts,
virtual local_actor_ptr exec(spawn_options opts,
init_callback init_cb,
void_function actor_behavior) = 0;
template<typename F, typename T, typename... Ts>
actor_ptr exec(spawn_options opts, init_callback cb,
local_actor_ptr exec(spawn_options opts, init_callback cb,
F f, T&& a0, Ts&&... as) {
return this->exec(opts, cb, std::bind(f, detail::fwd<T>(a0),
detail::fwd<Ts>(as)...));
......
This diff is collapsed.
......@@ -67,10 +67,6 @@ inline detail::empty_tuple* get_empty_tuple() {
return detail::singleton_manager::get_empty_tuple();
}
inline detail::decorated_names_map* get_decorated_names_map() {
return detail::singleton_manager::get_decorated_names_map();
}
} // namespace cppa
#endif // CPPA_SINGLETONS_HPP
......@@ -50,7 +50,8 @@ enum class spawn_options : int {
monitor_flag = 0x02,
detach_flag = 0x04,
hide_flag = 0x08,
blocking_api_flag = 0x10
blocking_api_flag = 0x10,
priority_aware_flag = 0x20
};
#endif
......@@ -58,6 +59,15 @@ enum class spawn_options : int {
namespace {
#endif
/**
* @brief Concatenates two {@link spawn_options}.
* @relates spawn_options
*/
constexpr spawn_options operator+(const spawn_options& lhs,
const spawn_options& rhs) {
return static_cast<spawn_options>(static_cast<int>(lhs) | static_cast<int>(rhs));
}
/**
* @brief Denotes default settings.
*/
......@@ -93,19 +103,17 @@ constexpr spawn_options hidden = spawn_options::hide_flag;
*/
constexpr spawn_options blocking_api = spawn_options::blocking_api_flag;
/**
* @brief Causes the new actor to evaluate message priorities.
* @note This implicitly causes the actor to run in its own thread.
*/
constexpr spawn_options priority_aware = spawn_options::priority_aware_flag
+ spawn_options::detach_flag;
#ifndef CPPA_DOCUMENTATION
} // namespace <anonymous>
#endif
/**
* @brief Concatenates two {@link spawn_options}.
* @relates spawn_options
*/
constexpr spawn_options operator+(const spawn_options& lhs,
const spawn_options& rhs) {
return static_cast<spawn_options>(static_cast<int>(lhs) | static_cast<int>(rhs));
}
/**
* @brief Checks wheter @p haystack contains @p needle.
* @relates spawn_options
......@@ -122,6 +130,14 @@ constexpr bool has_detach_flag(spawn_options opts) {
return has_spawn_option(opts, detached);
}
/**
* @brief Checks wheter the {@link priority_aware} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_priority_aware_flag(spawn_options opts) {
return has_spawn_option(opts, priority_aware);
}
/**
* @brief Checks wheter the {@link hidden} flag is set in @p opts.
* @relates spawn_options
......
......@@ -94,8 +94,7 @@ class stacked : public Base {
protected:
template<typename... Ts>
stacked(std::function<void()> fun, Ts&&... args)
: Base(std::forward<Ts>(args)...), m_behavior(std::move(fun)) { }
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
virtual void do_become(behavior&& bhvr, bool discard_old) override {
become_impl(std::move(bhvr), discard_old, message_id());
......
......@@ -58,21 +58,13 @@ namespace cppa {
class self_type;
class scheduler_helper;
class thread_mapped_actor;
template<>
struct has_blocking_receive<thread_mapped_actor> : std::true_type { };
/**
* @brief An actor using the blocking API running in its own thread.
* @extends local_actor
*/
class thread_mapped_actor : public extend<local_actor,thread_mapped_actor>::with<mailbox_based,stacked,threaded> {
friend class self_type; // needs access to cleanup()
friend class scheduler_helper; // needs access to mailbox
friend class detail::receive_policy; // needs access to await_message(), etc.
friend class detail::behavior_stack; // needs same access as receive_policy
class thread_mapped_actor : public extend<local_actor,thread_mapped_actor>::
with<mailbox_based,stacked,threaded> {
typedef combined_type super;
......@@ -84,11 +76,7 @@ class thread_mapped_actor : public extend<local_actor,thread_mapped_actor>::with
inline void initialized(bool value) { m_initialized = value; }
bool initialized() const;
protected:
void cleanup(std::uint32_t reason);
virtual bool initialized() const override;
private:
......
......@@ -56,8 +56,10 @@ class threaded : public Base {
public:
typedef std::chrono::high_resolution_clock::time_point timeout_type;
template<typename... Ts>
threaded(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
threaded(Ts&&... args) : Base(std::forward<Ts>(args)...), m_initialized(false) { }
inline void reset_timeout() { }
......@@ -71,32 +73,19 @@ class threaded : public Base {
inline bool waits_for_timeout(std::uint32_t) { return false; }
mailbox_element* pop() {
wait_for_data();
virtual mailbox_element* try_pop() {
return this->m_mailbox.try_pop();
}
inline mailbox_element* try_pop() {
return this->m_mailbox.try_pop();
mailbox_element* pop() {
wait_for_data();
return try_pop();
}
template<typename TimePoint>
mailbox_element* try_pop(const TimePoint& abs_time) {
inline mailbox_element* try_pop(const timeout_type& abs_time) {
return (timed_wait_for_data(abs_time)) ? try_pop() : nullptr;
}
bool push_back(mailbox_element* new_element) {
switch (this->m_mailbox.enqueue(new_element)) {
case intrusive::first_enqueued: {
lock_type guard(m_mtx);
m_cv.notify_one();
return true;
}
default: return true;
case intrusive::queue_closed: return false;
}
}
void run_detached() {
auto dthis = util::dptr<Subtype>(this);
dthis->init();
......@@ -104,19 +93,46 @@ class threaded : public Base {
dthis->on_exit();
}
inline void initialized(bool value) {
m_initialized = value;
}
virtual bool initialized() const override {
return m_initialized;
}
protected:
typedef threaded combined_type;
virtual void enqueue(const message_header& hdr, any_tuple msg) override {
void enqueue_impl(typename Base::mailbox_type& mbox,
const message_header& hdr,
any_tuple&& msg) {
auto ptr = this->new_mailbox_element(hdr, std::move(msg));
if (!push_back(ptr) && hdr.id.valid()) {
switch (mbox.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{this->exit_reason()};
f(hdr.sender, hdr.id);
}
break;
}
}
typedef std::chrono::high_resolution_clock::time_point timeout_type;
virtual void enqueue(const message_header& hdr, any_tuple msg) override {
enqueue_impl(this->m_mailbox, hdr, std::move(msg));
}
virtual bool chained_enqueue(const message_header& hdr, any_tuple msg) override {
enqueue(hdr, std::move(msg));
return false;
}
timeout_type init_timeout(const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
......@@ -132,11 +148,15 @@ class threaded : public Base {
return try_pop(abs_time);
}
virtual bool mailbox_empty() {
return this->m_mailbox.empty();
}
bool timed_wait_for_data(const timeout_type& abs_time) {
CPPA_REQUIRE(not this->m_mailbox.closed());
if (this->m_mailbox.empty()) {
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (this->m_mailbox.empty()) {
while (mailbox_empty()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) {
return false;
}
......@@ -146,15 +166,19 @@ class threaded : public Base {
}
void wait_for_data() {
if (this->m_mailbox.empty()) {
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (this->m_mailbox.empty()) m_cv.wait(guard);
while (mailbox_empty()) m_cv.wait(guard);
}
}
std::mutex m_mtx;
std::condition_variable m_cv;
private:
bool m_initialized;
};
} // namespace cppa
......
......@@ -76,9 +76,16 @@ inline std::string to_string(const group_ptr& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const channel_ptr& what) {
return detail::to_string_impl(what);
}
// implemented in process_information.cpp
std::string to_string(const process_information& what);
// implemented in process_information.cpp
std::string to_string(const process_information_ptr& what);
inline std::string to_string(const object& what) {
return detail::to_string_impl(what.value(), what.type());
}
......
......@@ -186,12 +186,6 @@ class uniform_type_info {
*/
static std::vector<const uniform_type_info*> instances();
/**
* @brief Get the internal @p libcppa name for this type.
* @returns A string describing the @p libcppa internal type name.
*/
inline const std::string& name() const { return m_name; }
/**
* @brief Creates an object of this type.
*/
......@@ -202,6 +196,12 @@ class uniform_type_info {
*/
object deserialize(deserializer* source) const;
/**
* @brief Get the internal @p libcppa name for this type.
* @returns A string describing the @p libcppa internal type name.
*/
virtual const char* name() const = 0;
/**
* @brief Determines if this uniform_type_info describes the same
* type than @p tinfo.
......@@ -234,23 +234,9 @@ class uniform_type_info {
*/
virtual void deserialize(void* instance, deserializer* source) const = 0;
/**
* @brief Checks wheter <tt>src->seek_object()</tt>
* returns @p tname and throws an exception if not.
* @throws std::logic_error
*/
static void assert_type_name(deserializer* src, const std::string& tname);
protected:
/**
* @brief Checks wheter <tt>src->seek_object()</tt>
* returns {@link name()} and throws an exception if not.
* @throws std::logic_error
*/
void assert_type_name(deserializer* src) const;
uniform_type_info(const std::string& uniform_name);
uniform_type_info() = default;
/**
* @brief Casts @p instance to the native type and deletes it.
......@@ -270,10 +256,6 @@ class uniform_type_info {
*/
virtual void* new_instance(const void* instance = nullptr) const = 0;
private:
std::string m_name;
};
/**
......
......@@ -31,8 +31,11 @@
#ifndef CPPA_ABSTRACT_UNIFORM_TYPE_INFO_HPP
#define CPPA_ABSTRACT_UNIFORM_TYPE_INFO_HPP
#include "cppa/deserializer.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
namespace cppa { namespace util {
......@@ -43,19 +46,23 @@ namespace cppa { namespace util {
template<typename T>
class abstract_uniform_type_info : public uniform_type_info {
inline static const T& deref(const void* ptr) {
return *reinterpret_cast<const T*>(ptr);
public:
bool equals(const std::type_info& tinfo) const {
return typeid(T) == tinfo;
}
inline static T& deref(void* ptr) {
return *reinterpret_cast<T*>(ptr);
const char* name() const {
return m_name.c_str();
}
protected:
abstract_uniform_type_info(const std::string& uname
= detail::to_uniform_name(typeid(T)))
: uniform_type_info(uname) {
abstract_uniform_type_info() {
auto uname = detail::to_uniform_name<T>();
auto cname = detail::mapped_name_by_decorated_name(uname.c_str());
if (cname == uname.c_str()) m_name = std::move(uname);
else m_name = cname;
}
bool equals(const void* lhs, const void* rhs) const {
......@@ -70,14 +77,32 @@ class abstract_uniform_type_info : public uniform_type_info {
delete reinterpret_cast<T*>(instance);
}
public:
void assert_type_name(deserializer* source) const {
auto tname = source->seek_object();
if (tname != name()) {
std::string error_msg = "wrong type name found; expected \"";
error_msg += name();
error_msg += "\", found \"";
error_msg += tname;
error_msg += "\"";
throw std::logic_error(std::move(error_msg));
}
}
bool equals(const std::type_info& tinfo) const {
return typeid(T) == tinfo;
private:
static inline const T& deref(const void* ptr) {
return *reinterpret_cast<const T*>(ptr);
}
static inline T& deref(void* ptr) {
return *reinterpret_cast<T*>(ptr);
}
std::string m_name;
};
} }
} } // namespace cppa::util
#endif // CPPA_ABSTRACT_UNIFORM_TYPE_INFO_HPP
......@@ -340,6 +340,18 @@ struct tl_zip_with_index<empty_type_list> {
typedef empty_type_list type;
};
// int index_of(list, type)
template<class List, typename T>
struct tl_index_of {
static constexpr size_t value = tl_index_of<typename tl_tail<List>::type,T>::value;
};
template<size_t N, typename T, typename... Ts>
struct tl_index_of<type_list<type_pair<std::integral_constant<size_t,N>,T>,Ts...>,T> {
static constexpr size_t value = N;
};
// list reverse()
template<class From, typename... Elements>
......
......@@ -35,6 +35,7 @@
#include <atomic>
#include <stdexcept>
#include "cppa/send.hpp"
#include "cppa/actor.hpp"
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
......@@ -60,22 +61,14 @@ actor::actor(actor_id aid)
actor::actor()
: m_id(get_actor_registry()->next_id()), m_is_proxy(false)
, m_exit_reason(exit_reason::not_exited) {
CPPA_LOG_INFO("spawned new actor with ID " << id()
<< ", class = " << CPPA_CLASS_NAME);
}
bool actor::chained_enqueue(const message_header& hdr, any_tuple msg) {
enqueue(hdr, std::move(msg));
return false;
}
, m_exit_reason(exit_reason::not_exited) { }
bool actor::link_to_impl(const actor_ptr& other) {
if (other && other != this) {
guard_type guard(m_mtx);
guard_type guard{m_mtx};
// send exit message if already exited
if (exited()) {
other->enqueue(this, make_any_tuple(atom("EXIT"), exit_reason()));
send_as(this, other, atom("EXIT"), exit_reason());
}
// add link if not already linked to other
// (checked by establish_backlink)
......@@ -89,12 +82,12 @@ bool actor::link_to_impl(const actor_ptr& other) {
bool actor::attach(attachable_ptr ptr) {
if (ptr == nullptr) {
guard_type guard(m_mtx);
guard_type guard{m_mtx};
return m_exit_reason == exit_reason::not_exited;
}
std::uint32_t reason;
{ // lifetime scope of guard
guard_type guard(m_mtx);
guard_type guard{m_mtx};
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
m_attachables.push_back(std::move(ptr));
......@@ -108,7 +101,7 @@ bool actor::attach(attachable_ptr ptr) {
void actor::detach(const attachable::token& what) {
attachable_ptr ptr;
{ // lifetime scope of guard
guard_type guard(m_mtx);
guard_type guard{m_mtx};
auto end = m_attachables.end();
auto i = std::find_if(
m_attachables.begin(), end,
......@@ -131,7 +124,7 @@ void actor::unlink_from(const actor_ptr& other) {
bool actor::remove_backlink(const actor_ptr& other) {
if (other && other != this) {
guard_type guard(m_mtx);
guard_type guard{m_mtx};
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i != m_links.end()) {
m_links.erase(i);
......@@ -144,7 +137,7 @@ bool actor::remove_backlink(const actor_ptr& other) {
bool actor::establish_backlink(const actor_ptr& other) {
std::uint32_t reason = exit_reason::not_exited;
if (other && other != this) {
guard_type guard(m_mtx);
guard_type guard{m_mtx};
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
auto i = std::find(m_links.begin(), m_links.end(), other);
......@@ -156,13 +149,13 @@ bool actor::establish_backlink(const actor_ptr& other) {
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
other->enqueue(this, make_any_tuple(atom("EXIT"), reason));
send_as(this, other, make_any_tuple(atom("EXIT"), reason));
}
return false;
}
bool actor::unlink_from_impl(const actor_ptr& other) {
guard_type guard(m_mtx);
guard_type guard{m_mtx};
// remove_backlink returns true if this actor is linked to other
if (other && !exited() && other->remove_backlink(this)) {
auto i = std::find(m_links.begin(), m_links.end(), other);
......@@ -174,12 +167,14 @@ bool actor::unlink_from_impl(const actor_ptr& other) {
}
void actor::cleanup(std::uint32_t reason) {
// log as 'actor'
CPPA_LOGM_TRACE("cppa::actor", CPPA_ARG(m_id) << ", " << CPPA_ARG(reason));
CPPA_REQUIRE(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
decltype(m_links) mlinks;
decltype(m_attachables) mattachables;
{ // lifetime scope of guard
guard_type guard(m_mtx);
guard_type guard{m_mtx};
if (m_exit_reason != exit_reason::not_exited) {
// already exited
return;
......@@ -191,9 +186,16 @@ void actor::cleanup(std::uint32_t reason) {
m_links.clear();
m_attachables.clear();
}
CPPA_LOGC_INFO_IF(not is_proxy(), "cppa::actor", __func__,
"actor with ID " << m_id << " had " << mlinks.size()
<< " links and " << mattachables.size()
<< " attached functors; exit reason = " << reason
<< ", class = " << detail::demangle(typeid(*this)));
// send exit messages
auto msg = make_any_tuple(atom("EXIT"), reason);
for (actor_ptr& aptr : mlinks) {
aptr->enqueue(this, make_any_tuple(atom("EXIT"), reason));
message_header hdr{this, aptr, message_priority::high};
hdr.deliver(msg);
}
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
......
......@@ -32,6 +32,7 @@
#include <limits>
#include <stdexcept>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/attachable.hpp"
#include "cppa/exit_reason.hpp"
......@@ -58,7 +59,7 @@ actor_registry::value_type actor_registry::get_entry(actor_id key) const {
if (i != m_entries.end()) {
return i->second;
}
CPPA_LOG_DEBUG("no cache entry found for " << CPPA_ARG(key));
CPPA_LOGMF(CPPA_DEBUG, self, "key not found: " << key);
return {nullptr, exit_reason::not_exited};
}
......@@ -76,7 +77,7 @@ void actor_registry::put(actor_id key, const actor_ptr& value) {
}
}
if (add_attachable) {
CPPA_LOG_INFO("added actor with ID " << key);
CPPA_LOGMF(CPPA_INFO, self, "added actor with ID " << key);
struct eraser : attachable {
actor_id m_id;
actor_registry* m_registry;
......@@ -97,7 +98,7 @@ void actor_registry::erase(actor_id key, std::uint32_t reason) {
auto i = m_entries.find(key);
if (i != m_entries.end()) {
auto& entry = i->second;
CPPA_LOG_INFO("erased actor with ID " << key
CPPA_LOGMF(CPPA_INFO, self, "erased actor with ID " << key
<< ", reason " << reason);
entry.first = nullptr;
entry.second = reason;
......@@ -109,7 +110,11 @@ std::uint32_t actor_registry::next_id() {
}
void actor_registry::inc_running() {
# if CPPA_LOG_LEVEL >= CPPA_DEBUG
CPPA_LOG_DEBUG("new value = " << ++m_running);
# else
++m_running;
# endif
}
size_t actor_registry::running() const {
......@@ -126,12 +131,14 @@ void actor_registry::dec_running() {
std::unique_lock<std::mutex> guard(m_running_mtx);
m_running_cv.notify_all();
}
CPPA_LOG_DEBUG(CPPA_ARG(new_val));
}
void actor_registry::await_running_count_equal(size_t expected) {
CPPA_LOG_TRACE(CPPA_ARG(expected));
std::unique_lock<std::mutex> guard(m_running_mtx);
while (m_running.load() != expected) {
CPPA_LOGMF(CPPA_TRACE, self.unchecked(), CPPA_ARG(expected));
std::unique_lock<std::mutex> guard{m_running_mtx};
while (m_running != expected) {
CPPA_LOG_DEBUG("count = " << m_running.load());
m_running_cv.wait(guard);
}
}
......
......@@ -38,6 +38,7 @@
#include <stdexcept>
#include <type_traits>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/binary_deserializer.hpp"
......@@ -51,7 +52,7 @@ typedef const char* iterator;
inline void range_check(iterator begin, iterator end, size_t read_size) {
if ((begin + read_size) > end) {
CPPA_LOGF_ERROR("range_check failed");
CPPA_LOGF(CPPA_ERROR, self, "range_check failed");
throw out_of_range("binary_deserializer::read_range()");
}
}
......@@ -153,7 +154,7 @@ void binary_deserializer::begin_object(const string&) { }
void binary_deserializer::end_object() { }
size_t binary_deserializer::begin_sequence() {
CPPA_LOG_TRACE("");
CPPA_LOGMF(CPPA_TRACE, self, "");
static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)");
uint32_t result;
......
......@@ -29,9 +29,15 @@
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
namespace cppa {
channel::~channel() { }
bool channel::chained_enqueue(const message_header& hdr, any_tuple msg) {
enqueue(hdr, std::move(msg));
return false;
}
} // namespace cppa
......@@ -39,8 +39,10 @@
namespace cppa {
context_switching_actor::context_switching_actor(std::function<void()> fun)
: super(std::move(fun), actor_state::ready, true)
, m_fiber(&context_switching_actor::trampoline, this) { }
: super(actor_state::ready, true)
, m_fiber(&context_switching_actor::trampoline, this) {
set_behavior(std::move(fun));
}
auto context_switching_actor::init_timeout(const util::duration& tout) -> timeout_type {
// request explicit timeout message
......@@ -94,7 +96,7 @@ scheduled_actor_type context_switching_actor::impl_type() {
}
resume_result context_switching_actor::resume(util::fiber* from, actor_ptr& next_job) {
CPPA_LOG_TRACE("id = " << id() << ", state = " << static_cast<int>(state()));
CPPA_LOGMF(CPPA_TRACE, this, "state = " << static_cast<int>(state()));
CPPA_REQUIRE(from != nullptr);
CPPA_REQUIRE(next_job == nullptr);
using namespace detail;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/>. *
\******************************************************************************/
#include <limits>
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/decorated_names_map.hpp"
using namespace std;
namespace {
using namespace cppa;
using namespace cppa::detail;
constexpr const char* mapped_int_names[][2] = {
{ nullptr, nullptr }, // sizeof 0-> invalid
{ "@i8", "@u8" }, // sizeof 1 -> signed / unsigned int8
{ "@i16", "@u16" }, // sizeof 2 -> signed / unsigned int16
{ nullptr, nullptr }, // sizeof 3-> invalid
{ "@i32", "@u32" }, // sizeof 4 -> signed / unsigned int32
{ nullptr, nullptr }, // sizeof 5-> invalid
{ nullptr, nullptr }, // sizeof 6-> invalid
{ nullptr, nullptr }, // sizeof 7-> invalid
{ "@i64", "@u64" } // sizeof 8 -> signed / unsigned int64
};
template<typename T>
constexpr size_t sign_index() {
static_assert(numeric_limits<T>::is_integer, "T is not an integer");
return numeric_limits<T>::is_signed ? 0 : 1;
}
template<typename T>
inline string demangled() {
return demangle(typeid(T));
}
template<typename T>
constexpr const char* mapped_int_name() {
return mapped_int_names[sizeof(T)][sign_index<T>()];
}
template<typename T>
struct is_signed : integral_constant<bool, numeric_limits<T>::is_signed> { };
} // namespace <anonymous>
namespace cppa { namespace detail {
decorated_names_map::decorated_names_map() {
map<string, string> tmp = {
// integer types
{ demangled<char>(), mapped_int_name<char>() },
{ demangled<signed char>(), mapped_int_name<signed char>() },
{ demangled<unsigned char>(), mapped_int_name<unsigned char>() },
{ demangled<short>(), mapped_int_name<short>() },
{ demangled<signed short>(), mapped_int_name<signed short>() },
{ demangled<unsigned short>(), mapped_int_name<unsigned short>() },
{ demangled<short int>(), mapped_int_name<short int>() },
{ demangled<signed short int>(), mapped_int_name<signed short int>() },
{ demangled<unsigned short int>(), mapped_int_name<unsigned short int>()},
{ demangled<int>(), mapped_int_name<int>() },
{ demangled<signed int>(), mapped_int_name<signed int>() },
{ demangled<unsigned int>(), mapped_int_name<unsigned int>() },
{ demangled<long int>(), mapped_int_name<long int>() },
{ demangled<signed long int>(), mapped_int_name<signed long int>() },
{ demangled<unsigned long int>(), mapped_int_name<unsigned long int>() },
{ demangled<long>(), mapped_int_name<long>() },
{ demangled<signed long>(), mapped_int_name<signed long>() },
{ demangled<unsigned long>(), mapped_int_name<unsigned long>() },
{ demangled<long long>(), mapped_int_name<long long>() },
{ demangled<signed long long>(), mapped_int_name<signed long long>() },
{ demangled<unsigned long long>(), mapped_int_name<unsigned long long>()},
{ demangled<char16_t>(), mapped_int_name<char16_t>() },
{ demangled<char32_t>(), mapped_int_name<char32_t>() },
{ "cppa::atom_value", "@atom" },
{ "cppa::any_tuple", "@<>" },
{ "cppa::message_header", "@hdr" },
{ "cppa::intrusive_ptr<cppa::actor>", "@actor" },
{ "cppa::intrusive_ptr<cppa::group>" ,"@group" },
{ "cppa::intrusive_ptr<cppa::channel>", "@channel" },
{ "cppa::intrusive_ptr<cppa::process_information>", "@process_info" },
{ "std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>", "@str" },
{ "std::basic_string<@u16,std::char_traits<@u16>,std::allocator<@u16>>", "@u16str" },
{ "std::basic_string<@u32,std::char_traits<@u32>,std::allocator<@u32>>", "@u32str"},
{ "std::map<@str,@str,std::less<@str>,std::allocator<std::pair<const @str,@str>>>", "@strmap" },
{ "std::string", "@str" }, // GCC
{ "cppa::util::void_type", "@0" }
};
m_map = move(tmp);
}
const string& decorated_names_map::decorate(const string& what) const {
auto i = m_map.find(what);
return (i != m_map.end()) ? i->second : what;
}
} } // namespace cppa::detail
......@@ -56,7 +56,7 @@ atom_value default_actor_addressing::technology_id() const {
void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
CPPA_REQUIRE(sink != nullptr);
if (ptr == nullptr) {
CPPA_LOG_DEBUG("serialized nullptr");
CPPA_LOGMF(CPPA_DEBUG, self, "serialized nullptr");
sink->begin_object("@0");
sink->end_object();
}
......@@ -69,9 +69,7 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
if (ptr->is_proxy()) {
auto dptr = ptr.downcast<default_actor_proxy>();
if (dptr) pinf = dptr->process_info();
else {
CPPA_LOG_ERROR("ptr is not a default_actor_proxy instance");
}
else CPPA_LOGMF(CPPA_ERROR, self, "downcast failed");
}
sink->begin_object("@actor");
sink->write_value(ptr->id());
......@@ -86,7 +84,7 @@ actor_ptr default_actor_addressing::read(deserializer* source) {
CPPA_REQUIRE(source != nullptr);
auto cname = source->seek_object();
if (cname == "@0") {
CPPA_LOG_DEBUG("deserialized nullptr");
CPPA_LOGMF(CPPA_DEBUG, self, "deserialized nullptr");
source->begin_object("@0");
source->end_object();
return nullptr;
......@@ -123,9 +121,9 @@ actor_ptr default_actor_addressing::get(const process_information& inf,
auto i = submap.find(aid);
if (i != submap.end()) {
auto result = i->second.promote();
CPPA_LOG_INFO_IF(!result, "proxy instance expired; "
<< CPPA_TARG(inf, to_string) << ", "
<< CPPA_ARG(aid));
CPPA_LOGMF_IF(!result, CPPA_INFO, self, "proxy instance expired; "
<< CPPA_TARG(inf, to_string) << ", "<< CPPA_ARG(aid));
if (!result) submap.erase(i);
return result;
}
return nullptr;
......@@ -145,15 +143,8 @@ void default_actor_addressing::put(const process_information& node,
aid));
}
else {
if (i->second.expired()) {
CPPA_LOG_INFO("removed expired weak pointer");
submap.erase(i);
put(node, aid, proxy);
}
else {
CPPA_LOG_ERROR("a proxy for " << aid << ":" << to_string(node)
<< " already exists");
}
CPPA_LOGMF(CPPA_ERROR, self, "proxy for " << aid << ":"
<< to_string(node) << " already exists");
}
}
......@@ -162,7 +153,7 @@ actor_ptr default_actor_addressing::get_or_put(const process_information& inf,
actor_id aid) {
auto result = get(inf, aid);
if (result == nullptr) {
CPPA_LOG_INFO("created new proxy instance; "
CPPA_LOGMF(CPPA_INFO, self, "created new proxy instance; "
<< CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
auto ptr = make_counted<default_actor_proxy>(aid, new process_information(inf), m_parent);
put(inf, aid, ptr);
......@@ -176,12 +167,12 @@ auto default_actor_addressing::proxies(process_information& i) -> proxy_map& {
}
void default_actor_addressing::erase(process_information& inf) {
CPPA_LOG_TRACE("inf = " << to_string(inf));
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string));
m_proxies.erase(inf);
}
void default_actor_addressing::erase(process_information& inf, actor_id aid) {
CPPA_LOG_TRACE("inf = " << to_string(inf) << ", aid = " << aid);
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
auto i = m_proxies.find(inf);
if (i != m_proxies.end()) {
i->second.erase(aid);
......
......@@ -52,12 +52,18 @@ sync_request_info::sync_request_info(actor_ptr sptr, message_id id)
default_actor_proxy::default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
const default_protocol_ptr& parent)
: super(mid), m_proto(parent), m_pinf(pinfo) { }
: super(mid), m_proto(parent), m_pinf(pinfo) {
CPPA_REQUIRE(parent != nullptr);
CPPA_LOG_INFO(CPPA_ARG(mid) << ", " << CPPA_TARG(pinfo, to_string)
<< "protocol = " << detail::demangle(typeid(*parent)));
}
default_actor_proxy::~default_actor_proxy() {
auto aid = id();
auto aid = m_id;
auto node = m_pinf;
auto proto = m_proto;
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(m_pinf)
<< ", protocol = " << detail::demangle(typeid(*m_proto)));
proto->run_later([aid, node, proto] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"~default_actor_proxy$run_later",
......@@ -88,7 +94,14 @@ void default_actor_proxy::deliver(const message_header& hdr, any_tuple msg) {
}
void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE("");
CPPA_LOG_TRACE(CPPA_ARG(m_id) << ", " << CPPA_TSARG(hdr)
<< ", " << CPPA_TSARG(msg));
if (hdr.receiver != this) {
auto cpy = hdr;
cpy.receiver = this;
forward_msg(cpy, std::move(msg));
return;
}
if (hdr.sender && hdr.id.is_request()) {
switch (m_pending_requests.enqueue(new_req_info(hdr.sender, hdr.id))) {
case intrusive::queue_closed: {
......@@ -116,15 +129,7 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
}
void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_TARG(hdr.sender, to_string) << ", "
<< CPPA_MARG(hdr.id, integer_value) << ", "
<< CPPA_TARG(msg, to_string));
// make sure header is complete, i.e., hdr.receiver == this, because
// the protocol relies on this information
if (hdr.receiver != this) {
enqueue({hdr.sender, this, hdr.id}, std::move(msg));
return;
}
CPPA_LOG_TRACE(CPPA_TARG(hdr, to_string) << ", " << CPPA_TARG(msg, to_string));
auto& arr = detail::static_types_array<atom_value,uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
......@@ -143,9 +148,8 @@ void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
f(e.sender.get(), e.mid);
});
});
return;
}
forward_msg(hdr, move(msg));
else forward_msg(hdr, move(msg));
}
void default_actor_proxy::link_to(const intrusive_ptr<actor>& other) {
......
......@@ -32,6 +32,7 @@
#include <cstdint>
#include "cppa/on.hpp"
#include "cppa/send.hpp"
#include "cppa/actor.hpp"
#include "cppa/match.hpp"
#include "cppa/logging.hpp"
......@@ -123,7 +124,7 @@ continue_reading_result default_peer::continue_reading() {
<< std::endl;
return read_failure;
}
CPPA_LOG_DEBUG("read process info: " << to_string(*m_node));
CPPA_LOGMF(CPPA_DEBUG, self, "read process info: " << to_string(*m_node));
m_parent->register_peer(*m_node, this);
// initialization done
m_state = wait_for_msg_size;
......@@ -149,12 +150,12 @@ continue_reading_result default_peer::continue_reading() {
m_meta_msg->deserialize(&msg, &bd);
}
catch (exception& e) {
CPPA_LOG_ERROR("exception during read_message: "
CPPA_LOGMF(CPPA_ERROR, self, "exception during read_message: "
<< detail::demangle(typeid(e))
<< ", what(): " << e.what());
return read_failure;
}
CPPA_LOG_DEBUG("deserialized: " << to_string(hdr) << " " << to_string(msg));
CPPA_LOGMF(CPPA_DEBUG, self, "deserialized: " << to_string(hdr) << " " << to_string(msg));
//DEBUG("<-- " << to_string(msg));
match(msg) (
// monitor messages are sent automatically whenever
......@@ -193,22 +194,22 @@ void default_peer::monitor(const actor_ptr&,
actor_id aid) {
CPPA_LOG_TRACE(CPPA_MARG(node, get) << ", " << CPPA_ARG(aid));
if (!node) {
CPPA_LOG_ERROR("received MONITOR from invalid peer");
CPPA_LOGMF(CPPA_ERROR, self, "received MONITOR from invalid peer");
return;
}
auto entry = get_actor_registry()->get_entry(aid);
auto pself = process_information::get();
if (*node == *pself) {
CPPA_LOG_ERROR("received 'MONITOR' from pself");
CPPA_LOGMF(CPPA_ERROR, self, "received 'MONITOR' from pself");
}
else if (entry.first == nullptr) {
if (entry.second == exit_reason::not_exited) {
CPPA_LOG_ERROR("received MONITOR for unknown "
CPPA_LOGMF(CPPA_ERROR, self, "received MONITOR for unknown "
"actor id: " << aid);
}
else {
CPPA_LOG_DEBUG("received MONITOR for an actor "
CPPA_LOGMF(CPPA_DEBUG, self, "received MONITOR for an actor "
"that already finished "
"execution; reply KILL_PROXY");
// this actor already finished execution;
......@@ -218,7 +219,7 @@ void default_peer::monitor(const actor_ptr&,
}
}
else {
CPPA_LOG_DEBUG("attach functor to " << entry.first.get());
CPPA_LOGMF(CPPA_DEBUG, self, "attach functor to " << entry.first.get());
default_protocol_ptr proto = m_parent;
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
......@@ -236,25 +237,23 @@ void default_peer::kill_proxy(const actor_ptr& sender,
const process_information_ptr& node,
actor_id aid,
std::uint32_t reason) {
CPPA_LOG_TRACE(CPPA_MARG(sender, get)
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string)
<< ", " << CPPA_MARG(node, get)
<< ", " << CPPA_ARG(aid)
<< ", " << CPPA_ARG(reason));
if (!node) {
CPPA_LOG_ERROR("node = nullptr");
CPPA_LOGMF(CPPA_ERROR, self, "node = nullptr");
return;
}
if (sender != nullptr) {
CPPA_LOG_ERROR("sender != nullptr");
CPPA_LOGMF(CPPA_ERROR, self, "sender != nullptr");
return;
}
auto proxy = m_parent->addressing()->get(*node, aid);
if (proxy) {
CPPA_LOG_DEBUG("received KILL_PROXY for " << aid
CPPA_LOGMF(CPPA_DEBUG, self, "received KILL_PROXY for " << aid
<< ":" << to_string(*node));
proxy->enqueue(nullptr,
make_any_tuple(
atom("KILL_PROXY"), reason));
send_as(nullptr, proxy, atom("KILL_PROXY"), reason);
}
else {
CPPA_LOG_INFO("received KILL_PROXY message but "
......@@ -269,24 +268,6 @@ void default_peer::deliver(const message_header& hdr, any_tuple msg) {
hdr.sender.downcast<actor_proxy>()->deliver(hdr, std::move(msg));
}
else hdr.deliver(std::move(msg));
/*
auto receiver = hdr.receiver.get();
if (receiver) {
if (hdr.id.valid()) {
CPPA_LOG_DEBUG("sync message for actor " << receiver->id());
receiver->sync_enqueue(hdr.sender.get(), hdr.id, move(msg));
}
else {
CPPA_LOG_DEBUG("async message with "
<< (hdr.sender ? "" : "in")
<< "valid sender");
receiver->enqueue(hdr.sender.get(), move(msg));
}
}
else {
CPPA_LOG_ERROR("received message with invalid receiver");
}
*/
}
void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
......@@ -332,13 +313,13 @@ continue_writing_result default_peer::continue_writing() {
size_t written;
try { written = m_out->write_some(m_wr_buf.data(), m_wr_buf.size()); }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
disconnected();
return write_failure;
}
if (written != m_wr_buf.size()) {
CPPA_LOG_DEBUG("tried to write " << m_wr_buf.size() << "bytes, "
CPPA_LOGMF(CPPA_DEBUG, self, "tried to write " << m_wr_buf.size() << "bytes, "
<< "only " << written << " bytes written");
m_wr_buf.erase_leading(written);
return write_continue_later;
......@@ -346,7 +327,7 @@ continue_writing_result default_peer::continue_writing() {
else {
m_wr_buf.reset();
m_has_unwritten_data = false;
CPPA_LOG_DEBUG("write done, " << written << "bytes written");
CPPA_LOGMF(CPPA_DEBUG, self, "write done, " << written << "bytes written");
}
// try to write next message in queue
while (!m_has_unwritten_data && !queue().empty()) {
......@@ -374,19 +355,19 @@ void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
m_wr_buf.write(sizeof(uint32_t), &size, util::grow_if_needed);
try { bs << hdr << msg; }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e));
cerr << "*** exception in default_peer::enqueue; "
<< to_verbose_string(e)
<< endl;
return;
}
CPPA_LOG_DEBUG("serialized: " << to_string(hdr) << " " << to_string(msg));
CPPA_LOGMF(CPPA_DEBUG, self, "serialized: " << to_string(hdr) << " " << to_string(msg));
size = (m_wr_buf.size() - before) - sizeof(std::uint32_t);
// update size in buffer
memcpy(m_wr_buf.data() + before, &size, sizeof(std::uint32_t));
CPPA_LOG_DEBUG_IF(m_has_unwritten_data, "still registered for writing");
if (!m_has_unwritten_data) {
CPPA_LOG_DEBUG("register for writing");
CPPA_LOGMF(CPPA_DEBUG, self, "register for writing");
m_has_unwritten_data = true;
m_parent->continue_writer(this);
}
......
......@@ -136,10 +136,10 @@ default_peer_ptr default_protocol::get_peer(const process_information& n) {
CPPA_LOG_TRACE("n = " << to_string(n));
auto i = m_peers.find(n);
if (i != m_peers.end()) {
CPPA_LOG_DEBUG("result = " << i->second.impl.get());
CPPA_LOGMF(CPPA_DEBUG, self, "result = " << i->second.impl.get());
return i->second.impl;
}
CPPA_LOG_DEBUG("result = nullptr");
CPPA_LOGMF(CPPA_DEBUG, self, "result = nullptr");
return nullptr;
}
......@@ -192,8 +192,8 @@ actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
auto pinfptr = make_counted<process_information>(peer_pid, peer_node_id);
if (*pinf == *pinfptr) {
// dude, this is not a remote actor, it's a local actor!
CPPA_LOG_ERROR("remote_actor() called to access a local actor");
# ifndef CPPA_DEBUG
CPPA_LOGMF(CPPA_ERROR, self, "remote_actor() called to access a local actor");
# ifndef CPPA_DEBUG_MODE
std::cerr << "*** warning: remote_actor() called to access a local actor\n"
<< std::flush;
# endif
......
......@@ -84,7 +84,7 @@ class default_scheduled_actor : public event_based_actor {
};
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> fun) {
return detail::memory::create<default_scheduled_actor>(std::move(fun));
return make_counted<default_scheduled_actor>(std::move(fun));
}
event_based_actor::event_based_actor(actor_state st) : super(st, true) { }
......@@ -119,7 +119,7 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
//e = m_mailbox.try_pop();
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOG_DEBUG("no more element in mailbox; going to block");
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
next_job.swap(m_chained_actor);
set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
......@@ -131,20 +131,20 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
next_job.swap(m_chained_actor);
CPPA_LOG_DEBUG("switched back to ready: "
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
break;
case actor_state::blocked:
CPPA_LOG_DEBUG("set state successfully to blocked");
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked");
// done setting actor to blocked
return resume_result::actor_blocked;
default:
CPPA_LOG_ERROR("invalid state");
CPPA_LOGMF(CPPA_ERROR, self, "invalid state");
CPPA_CRITICAL("invalid state");
};
}
else {
CPPA_LOG_DEBUG("switched back to ready: "
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"mailbox can fetch more");
set_state(actor_state::ready);
CPPA_REQUIRE(m_chained_actor == nullptr);
......@@ -152,14 +152,14 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
}
}
else {
CPPA_LOG_DEBUG("try to invoke message: " << to_string(e->msg));
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()) {
CPPA_LOG_DEBUG("behavior stack empty");
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
done_cb();
return resume_result::actor_done;
}
......
......@@ -36,6 +36,7 @@
#include <condition_variable>
#include "cppa/cppa.hpp"
#include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
......@@ -67,18 +68,23 @@ class local_group : public group {
public:
void send_all_subscribers(const actor_ptr& sender, const any_tuple& msg) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< CPPA_TARG(msg, to_string));
shared_guard guard(m_mtx);
for (auto& s : m_subscribers) {
s->enqueue(sender, msg);
send_tuple_as(sender, s, msg);
}
}
void enqueue(const message_header& hdr, any_tuple msg) override {
CPPA_LOG_TRACE(CPPA_TARG(hdr, to_string) << ", "
<< CPPA_TARG(msg, to_string));
send_all_subscribers(hdr.sender, msg);
m_broker->enqueue(hdr, move(msg));
}
pair<bool,size_t> add_subscriber(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TARG(who, to_string));
exclusive_guard guard(m_mtx);
if (m_subscribers.insert(who).second) {
return {true, m_subscribers.size()};
......@@ -87,12 +93,14 @@ class local_group : public group {
}
pair<bool,size_t> erase_subscriber(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TARG(who, to_string));
exclusive_guard guard(m_mtx);
auto erased_one = m_subscribers.erase(who) > 0;
return {erased_one, m_subscribers.size()};
auto success = m_subscribers.erase(who) > 0;
return {success, m_subscribers.size()};
}
group::subscription subscribe(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TARG(who, to_string));
if (add_subscriber(who).first) {
return {who, this};
}
......@@ -100,6 +108,7 @@ class local_group : public group {
}
void unsubscribe(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TARG(who, to_string));
erase_subscriber(who);
}
......@@ -130,27 +139,38 @@ class local_broker : public event_based_actor {
void init() {
become (
on(atom("JOIN"), arg_match) >> [=](const actor_ptr& other) {
CPPA_LOGC_TRACE("cppa::local_broker", "init$JOIN",
CPPA_TARG(other, to_string));
if (other && m_acquaintances.insert(other).second) {
monitor(other);
}
},
on(atom("LEAVE"), arg_match) >> [=](const actor_ptr& other) {
CPPA_LOGC_TRACE("cppa::local_broker", "init$LEAVE",
CPPA_TARG(other, to_string));
if (other && m_acquaintances.erase(other) > 0) {
demonitor(other);
}
},
on(atom("FORWARD"), arg_match) >> [=](const any_tuple& what) {
CPPA_LOGC_TRACE("cppa::local_broker", "init$FORWARD",
CPPA_TARG(what, to_string));
// local forwarding
m_group->send_all_subscribers(last_sender().get(), what);
m_group->send_all_subscribers(last_sender(), what);
// forward to all acquaintances
send_to_acquaintances(what);
},
on<atom("DOWN"), uint32_t>() >> [=] {
on(atom("DOWN"), arg_match) >> [=](uint32_t) {
actor_ptr other = last_sender();
CPPA_LOGC_TRACE("cppa::local_broker", "init$DOWN",
CPPA_TARG(other, to_string));
if (other) m_acquaintances.erase(other);
},
others() >> [=] {
send_to_acquaintances(last_dequeued());
auto msg = last_dequeued();
CPPA_LOGC_TRACE("cppa::local_broker", "init$others",
CPPA_TARG(msg, to_string));
send_to_acquaintances(msg);
}
);
}
......@@ -159,9 +179,12 @@ class local_broker : public event_based_actor {
void send_to_acquaintances(const any_tuple& what) {
// send to all remote subscribers
auto sender = last_sender().get();
auto sender = last_sender();
CPPA_LOG_DEBUG("forward message to " << m_acquaintances.size()
<< " acquaintances; " << CPPA_TSARG(sender)
<< ", " << CPPA_TSARG(what));
for (auto& acquaintance : m_acquaintances) {
acquaintance->enqueue(sender, what);
acquaintance->enqueue({sender, acquaintance}, what);
}
}
......@@ -193,12 +216,12 @@ class local_group_proxy : public local_group {
}
group::subscription subscribe(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TSARG(who));
auto res = add_subscriber(who);
if (res.first) {
if (res.second == 1) {
// join the remote source
m_broker->enqueue(nullptr,
make_any_tuple(atom("JOIN"), m_proxy_broker));
send_as(nullptr, m_broker, atom("JOIN"), m_proxy_broker);
}
return {who, this};
}
......@@ -206,12 +229,12 @@ class local_group_proxy : public local_group {
}
void unsubscribe(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TSARG(who));
auto res = erase_subscriber(who);
if (res.first && res.second == 0) {
// leave the remote source,
// because there's no more subscriber on this node
m_broker->enqueue(nullptr,
make_any_tuple(atom("LEAVE"), m_proxy_broker));
send_as(nullptr, m_broker, atom("LEAVE"), m_proxy_broker);
}
}
......@@ -237,8 +260,7 @@ class proxy_broker : public event_based_actor {
void init() {
become (
others() >> [=] {
m_group->send_all_subscribers(last_sender().get(),
last_dequeued());
m_group->send_all_subscribers(last_sender(), last_dequeued());
}
);
}
......@@ -336,19 +358,24 @@ class remote_group : public group {
: super(parent, move(id)), m_decorated(decorated) { }
group::subscription subscribe(const channel_ptr& who) {
CPPA_LOG_TRACE(CPPA_TSARG(who));
return m_decorated->subscribe(who);
}
void unsubscribe(const channel_ptr&) { /* never called */ }
void unsubscribe(const channel_ptr&) {
CPPA_LOG_ERROR("should never be called");
}
void enqueue(const message_header& hdr, any_tuple msg) override {
CPPA_LOG_TRACE("");
m_decorated->enqueue(hdr, std::move(msg));
}
void serialize(serializer* sink);
void group_down() {
group_ptr _this(this);
CPPA_LOG_TRACE("");
group_ptr _this{this};
m_decorated->send_all_subscribers(nullptr,
make_any_tuple(atom("GROUP_DOWN"),
_this));
......@@ -374,7 +401,7 @@ class shared_map : public ref_counted {
lock_type guard(m_mtx);
auto i = m_instances.find(key);
if (i == m_instances.end()) {
m_worker->enqueue(nullptr, make_any_tuple(atom("FETCH"), key));
send_as(nullptr, m_worker, atom("FETCH"), key);
do {
m_cond.wait(guard);
} while ((i = m_instances.find(key)) == m_instances.end());
......
......@@ -28,6 +28,7 @@
\******************************************************************************/
#include <string>
#include "cppa/cppa.hpp"
#include "cppa/atom.hpp"
#include "cppa/logging.hpp"
......@@ -54,8 +55,8 @@ class down_observer : public attachable {
void actor_exited(std::uint32_t reason) {
if (m_observer) {
m_observer->enqueue(m_observed.get(),
make_any_tuple(atom("DOWN"), reason));
message_header hdr{m_observed, m_observer, message_priority::high};
hdr.deliver(make_any_tuple(atom("DOWN"), reason));
}
}
......@@ -69,11 +70,41 @@ class down_observer : public attachable {
};
constexpr const char* s_default_debug_name = "actor";
} // namespace <anonymous>
local_actor::local_actor(bool sflag)
: m_chaining(sflag), m_trap_exit(false)
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node) { }
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node) {
# ifdef CPPA_DEBUG_MODE
new (&m_debug_name) std::string (std::to_string(m_id) + "@local");
# endif // CPPA_DEBUG_MODE
}
local_actor::~local_actor() {
using std::string;
# ifdef CPPA_DEBUG_MODE
m_debug_name.~string();
# endif // CPPA_DEBUG_MODE
}
const char* local_actor::debug_name() const {
# ifdef CPPA_DEBUG_MODE
return m_debug_name.c_str();
# else // CPPA_DEBUG_MODE
return s_default_debug_name;
# endif // CPPA_DEBUG_MODE
}
void local_actor::debug_name(std::string str) {
# ifdef CPPA_DEBUG_MODE
m_debug_name = std::move(str);
# else // CPPA_DEBUG_MODE
CPPA_LOG_WARNING("unable to set debug name to " << str
<< " (compiled without debug mode enabled)");
# endif // CPPA_DEBUG_MODE
}
void local_actor::monitor(const actor_ptr& whom) {
if (whom) whom->attach(attachable_ptr{new down_observer(this, whom)});
......@@ -113,15 +144,15 @@ void local_actor::reply_message(any_tuple&& what) {
}
auto& id = m_current_node->mid;
if (id.valid() == false || id.is_response()) {
send_message(whom.get(), std::move(what));
send_tuple(whom, std::move(what));
}
else if (!id.is_answered()) {
if (chaining_enabled()) {
if (whom->chained_enqueue({this, id.response_id()}, std::move(what))) {
if (whom->chained_enqueue({this, whom, id.response_id()}, std::move(what))) {
chained_actor(whom);
}
}
else whom->enqueue({this, id.response_id()}, std::move(what));
else whom->enqueue({this, whom, id.response_id()}, std::move(what));
id.mark_as_answered();
}
}
......@@ -129,7 +160,7 @@ void local_actor::reply_message(any_tuple&& what) {
void local_actor::forward_message(const actor_ptr& new_receiver) {
if (new_receiver == nullptr) return;
auto& id = m_current_node->mid;
new_receiver->enqueue({last_sender(), id}, m_current_node->msg);
new_receiver->enqueue({last_sender(), new_receiver, id}, m_current_node->msg);
// treat this message as asynchronous message from now on
id = message_id{};
}
......@@ -141,15 +172,6 @@ response_handle local_actor::make_response_handle() {
return std::move(result);
}
message_id local_actor::send_timed_sync_message(const actor_ptr& whom,
const util::duration& rel_time,
any_tuple&& what) {
auto mid = this->send_sync_message(whom, std::move(what));
auto tmp = make_any_tuple(atom("TIMEOUT"));
get_scheduler()->delayed_reply(this, rel_time, mid, std::move(tmp));
return mid;
}
void local_actor::exec_behavior_stack() {
// default implementation does nothing
}
......
......@@ -30,6 +30,7 @@
#include <ctime>
#include <thread>
#include <cstring>
#include <fstream>
#include <algorithm>
......@@ -40,6 +41,7 @@
#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/intrusive/blocking_single_reader_queue.hpp"
......@@ -70,11 +72,11 @@ class logging_impl : public logging {
void initialize() {
m_thread = thread([this] { (*this)(); });
log("TRACE", "logging", "run", __FILE__, __LINE__, "ENTRY");
log("TRACE", "logging", "run", __FILE__, __LINE__, nullptr, "ENTRY");
}
void destroy() {
log("TRACE", "logging", "run", __FILE__, __LINE__, "EXIT");
log("TRACE", "logging", "run", __FILE__, __LINE__, nullptr, "EXIT");
// an empty string means: shut down
m_queue.push_back(new log_event{0, ""});
m_thread.join();
......@@ -84,7 +86,7 @@ class logging_impl : public logging {
void operator()() {
ostringstream fname;
fname << "libcppa_" << getpid() << "_" << time(0) << ".log";
fstream out(fname.str().c_str(), ios::out);
fstream out(fname.str().c_str(), ios::out | ios::app);
unique_ptr<log_event> event;
for (;;) {
event.reset(m_queue.pop());
......@@ -101,6 +103,7 @@ class logging_impl : public logging {
const char* function_name,
const char* c_full_file_name,
int line_num,
const actor_ptr& from,
const std::string& msg) {
string class_name = c_class_name;
replace_all(class_name, "::", ".");
......@@ -114,9 +117,25 @@ class logging_impl : public logging {
else file_name = string(i, full_file_name.end());
}
else file_name = move(full_file_name);
auto print_from = [&](ostream& oss) -> ostream& {
if (!from) {
if (strcmp(c_class_name, "logging") == 0) oss << "logging";
else oss << "null";
}
else if (from->is_proxy()) oss << to_string(from);
else {
# ifdef CPPA_DEBUG_MODE
oss << from.downcast<local_actor>()->debug_name();
# else // CPPA_DEBUG_MODE
oss << from->id() << "@local";
# endif // CPPA_DEBUG_MODE
}
return oss;
};
ostringstream line;
line << time(0) << " "
<< level << " "
<< level << " ";
print_from(line) << " "
<< this_thread::get_id() << " "
<< class_name << " "
<< function_name << " "
......@@ -139,16 +158,17 @@ logging::trace_helper::trace_helper(std::string class_name,
const char* fun_name,
const char* file_name,
int line_num,
actor_ptr ptr,
const std::string& msg)
: m_class(std::move(class_name)), m_fun_name(fun_name)
, m_file_name(file_name), m_line_num(line_num) {
, m_file_name(file_name), m_line_num(line_num), m_self(std::move(ptr)) {
get_logger()->log("TRACE", m_class.c_str(), fun_name,
file_name, line_num, "ENTRY " + msg);
file_name, line_num, m_self, "ENTRY " + msg);
}
logging::trace_helper::~trace_helper() {
get_logger()->log("TRACE", m_class.c_str(), m_fun_name,
m_file_name, m_line_num, "EXIT");
m_file_name, m_line_num, m_self, "EXIT");
}
logging::~logging() { }
......
......@@ -34,20 +34,28 @@
namespace cppa {
namespace { using actor_cref = const actor_ptr&; }
message_header::message_header(const std::nullptr_t&)
: priority(message_priority::normal) { }
message_header::message_header(actor* s) : sender(s) { }
message_header::message_header(const self_type&)
: sender(self), receiver(self), priority(message_priority::normal) { }
message_header::message_header(const self_type& s) : sender(s) { }
message_header::message_header(channel_ptr dest, message_id mid, message_priority prio)
: sender(self), receiver(dest), id(mid), priority(prio) { }
message_header::message_header(actor_cref s, message_priority p)
: sender(s), priority(p) { }
message_header::message_header(channel_ptr dest, message_priority prio)
: sender(self), receiver(dest), priority(prio) { }
message_header::message_header(actor_cref s, message_id mid, message_priority p)
: sender(s), id(mid), priority(p) { }
message_header::message_header(actor_ptr source,
channel_ptr dest,
message_id mid,
message_priority prio)
: sender(source), receiver(dest), id(mid), priority(prio) { }
message_header::message_header(actor_cref s, actor_cref r, message_id mid, message_priority p)
: sender(s), receiver(r), id(mid), priority(p) { }
message_header::message_header(actor_ptr source,
channel_ptr dest,
message_priority prio)
: sender(source), receiver(dest), priority(prio) { }
bool operator==(const message_header& lhs, const message_header& rhs) {
return lhs.sender == rhs.sender
......
......@@ -45,6 +45,7 @@
#include "cppa/actor_proxy.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/process_information.hpp"
......@@ -146,7 +147,7 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
CPPA_REQUIRE(m_pollset.size() == m_meta.size());
for (;;) {
auto presult = ::poll(m_pollset.data(), m_pollset.size(), -1);
CPPA_LOG_DEBUG("poll() on " << num_sockets()
CPPA_LOGMF(CPPA_DEBUG, self, "poll() on " << num_sockets()
<< " sockets returned " << presult);
if (presult < 0) {
switch (errno) {
......@@ -156,7 +157,7 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
break;
}
case ENOMEM: {
CPPA_LOG_ERROR("poll() failed for reason ENOMEM");
CPPA_LOGMF(CPPA_ERROR, self, "poll() failed for reason ENOMEM");
// there's not much we can do other than try again
// in hope someone releases memory
//this_thread::yield();
......@@ -186,7 +187,7 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
tmp.events = to_poll_bitmask(new_bitmask);
tmp.revents = 0;
m_pollset.insert(lb(fd), tmp);
CPPA_LOG_DEBUG("inserted new element");
CPPA_LOGMF(CPPA_DEBUG, self, "inserted new element");
break;
}
case fd_meta_event::erase: {
......@@ -196,7 +197,7 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOG_DEBUG("erased element");
CPPA_LOGMF(CPPA_DEBUG, self, "erased element");
m_pollset.erase(iter);
}
break;
......@@ -208,7 +209,7 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOG_DEBUG("updated bitmask");
CPPA_LOGMF(CPPA_DEBUG, self, "updated bitmask");
iter->events = to_poll_bitmask(new_bitmask);
}
break;
......@@ -281,10 +282,10 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
pair<event_iterator,event_iterator> poll() {
CPPA_REQUIRE(m_meta.empty() == false);
for (;;) {
CPPA_LOG_DEBUG("epoll_wait on " << num_sockets() << " sockets");
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait on " << num_sockets() << " sockets");
auto presult = epoll_wait(m_epollfd, m_events.data(),
(int) m_events.size(), -1);
CPPA_LOG_DEBUG("epoll_wait returned " << presult);
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait returned " << presult);
if (presult < 0) {
// try again unless critical error occured
presult = 0;
......@@ -349,18 +350,18 @@ class middleman_event_handler : public middleman_event_handler_base<middleman_ev
switch (errno) {
// supplied file descriptor is already registered
case EEXIST: {
CPPA_LOG_ERROR("file descriptor registered twice");
CPPA_LOGMF(CPPA_ERROR, self, "file descriptor registered twice");
break;
}
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance.
case ENOENT: {
CPPA_LOG_ERROR("cannot delete file descriptor "
CPPA_LOGMF(CPPA_ERROR, self, "cannot delete file descriptor "
"because it isn't registered");
break;
}
default: {
CPPA_LOG_ERROR(strerror(errno));
CPPA_LOGMF(CPPA_ERROR, self, strerror(errno));
perror("epoll_ctl() failed");
CPPA_CRITICAL("epoll_ctl() failed");
}
......@@ -411,7 +412,7 @@ class middleman_impl : public abstract_middleman {
void add_protocol(const protocol_ptr& impl) {
if (!impl) {
CPPA_LOG_ERROR("impl == nullptr");
CPPA_LOGMF(CPPA_ERROR, self, "impl == nullptr");
throw std::invalid_argument("impl == nullptr");
}
CPPA_LOG_TRACE("identifier = " << to_string(impl->identifier()));
......@@ -491,14 +492,14 @@ class middleman_overseer : public continuable_reader {
static constexpr size_t num_dummies = 64;
uint8_t dummies[num_dummies];
auto read_result = ::read(read_handle(), dummies, num_dummies);
CPPA_LOG_DEBUG("read " << read_result << " messages from queue");
CPPA_LOGMF(CPPA_DEBUG, self, "read " << read_result << " messages from queue");
if (read_result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// try again later
return read_continue_later;
}
else {
CPPA_LOG_ERROR("cannot read from pipe");
CPPA_LOGMF(CPPA_ERROR, self, "cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe");
}
}
......@@ -506,10 +507,10 @@ class middleman_overseer : public continuable_reader {
for (int i = 0; i < read_result; ++i) {
unique_ptr<middleman_event> msg(m_queue.try_pop());
if (!msg) {
CPPA_LOG_ERROR("nullptr dequeued");
CPPA_LOGMF(CPPA_ERROR, self, "nullptr dequeued");
CPPA_CRITICAL("nullptr dequeued");
}
CPPA_LOG_DEBUG("execute run_later functor");
CPPA_LOGMF(CPPA_DEBUG, self, "execute run_later functor");
(*msg)();
}
return read_continue_later;
......@@ -559,6 +560,11 @@ void abstract_middleman::stop_reader(const continuable_reader_ptr& ptr) {
}
void middleman_loop(middleman_impl* impl) {
# ifdef CPPA_LOG_LEVEL
auto mself = make_counted<thread_mapped_actor>();
scoped_self_setter sss(mself.get());
CPPA_SET_DEBUG_NAME("middleman");
# endif
middleman_event_handler* handler = &impl->m_handler;
CPPA_LOGF_TRACE("run middleman loop");
CPPA_LOGF_INFO("middleman runs at "
......
......@@ -80,14 +80,14 @@ struct command_dispatcher::worker {
if (err != CL_SUCCESS) {
ostringstream oss;
oss << "clFlush: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
}
catch (exception& e) {
ostringstream oss;
oss << "worker loop, e.what(): " << e.what();
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
}
......@@ -107,12 +107,12 @@ void command_dispatcher::worker_loop(command_dispatcher::worker* w) {
void command_dispatcher::supervisor_loop(command_dispatcher* scheduler,
job_queue* jq, command_ptr m_dummy) {
CPPA_LOGF_TRACE("");
unique_ptr<command_dispatcher::worker> worker;
worker.reset(new command_dispatcher::worker(scheduler, jq, m_dummy));
worker->start();
worker->m_thread.join();
worker.reset();
CPPA_LOG_TRACE("supervisor done");
}
void command_dispatcher::initialize() {
......@@ -128,13 +128,13 @@ void command_dispatcher::initialize() {
ostringstream oss;
oss << "clGetPlatformIDs (getting number of platforms): "
<< get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw logic_error(oss.str());
}
else if (number_of_platforms < 1) {
ostringstream oss;
oss << "clGetPlatformIDs: no platforms found.";
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw logic_error(oss.str());
}
......@@ -144,7 +144,7 @@ void command_dispatcher::initialize() {
ostringstream oss;
oss << "clGetPlatformIDs (getting platform ids): "
<< get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw logic_error(oss.str());
}
......@@ -162,7 +162,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) {
ostringstream oss;
oss << "clGetDeviceIDs: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
vector<cl_device_id> devices(num_devices);
......@@ -170,7 +170,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) {
ostringstream oss;
oss << "clGetDeviceIDs: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
......@@ -179,7 +179,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) {
ostringstream oss;
oss << "clCreateContext: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
......@@ -193,7 +193,7 @@ void command_dispatcher::initialize() {
char buf[buf_size];
err = clGetDeviceInfo(device.get(), CL_DEVICE_NAME, buf_size, buf, &return_size);
if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clGetDeviceInfo (CL_DEVICE_NAME): " << get_opencl_error(err));
CPPA_LOGMF(CPPA_ERROR, self, "clGetDeviceInfo (CL_DEVICE_NAME): " << get_opencl_error(err));
fill(buf, buf+buf_size, 0);
}
command_queue_ptr cmd_queue;
......@@ -202,7 +202,7 @@ void command_dispatcher::initialize() {
CL_QUEUE_PROFILING_ENABLE,
&err));
if (err != CL_SUCCESS) {
CPPA_LOG_DEBUG("Could not create command queue for device "
CPPA_LOGMF(CPPA_DEBUG, self, "Could not create command queue for device "
<< buf << ": " << get_opencl_error(err));
}
else {
......@@ -217,7 +217,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_GROUP_SIZE): "
<< get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
cl_uint max_work_item_dimensions = 0;
......@@ -231,7 +231,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_ITEM_DIMENSIONS): "
<< get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
dim_vec max_work_items_per_dim(max_work_item_dimensions);
......@@ -245,7 +245,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_ITEM_SIZES): "
<< get_opencl_error(err);
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
device_info dev_info{id,
......@@ -261,7 +261,7 @@ void command_dispatcher::initialize() {
ostringstream oss;
oss << "Could not create a command queue for "
<< "any of the present devices.";
CPPA_LOG_ERROR(oss.str());
CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str());
}
else {
......
......@@ -89,11 +89,11 @@ program program::create(const char* kernel_source) {
std::ostringstream oss;
oss << "clBuildProgram: " << get_opencl_error(err)
<< ", build log: " << build_log.data();
CPPA_LOGM_ERROR(detail::demangle<program>(), oss.str());
CPPA_LOGM_ERROR(detail::demangle<program>().c_str(), oss.str());
throw std::runtime_error(oss.str());
}
else {
# ifdef CPPA_DEBUG
# ifdef CPPA_DEBUG_MODE
device_ptr device_used(cppa::detail::singleton_manager::
get_command_dispatcher()->
m_devices.front().dev_id);
......@@ -119,7 +119,7 @@ program program::create(const char* kernel_source) {
build_log.data(),
nullptr);
build_log[ret_val_size] = '\0';
CPPA_LOG_DEBUG("clBuildProgram build log: "
CPPA_LOGM_DEBUG("program", "clBuildProgram build log: "
<< build_log.data());
# endif
}
......
......@@ -204,4 +204,13 @@ std::string to_string(const process_information& what) {
return oss.str();
}
std::string to_string(const process_information_ptr& what) {
std::ostringstream oss;
oss << "@process_info(";
if (!what) oss << "null";
else oss << to_string(*what);
oss << ")";
return oss.str();
}
} // namespace cppa
......@@ -105,7 +105,7 @@ bool scheduled_actor::enqueue_impl(actor_state next_state,
if (m_state.compare_exchange_weak(state, next_state)) {
CPPA_REQUIRE(m_scheduler != nullptr);
if (next_state == actor_state::ready) {
CPPA_LOG_DEBUG("enqueued actor with id " << id()
CPPA_LOGMF(CPPA_DEBUG, self, "enqueued actor with id " << id()
<< " to job queue");
m_scheduler->enqueue(this);
return false;
......
......@@ -108,8 +108,8 @@ class scheduler_helper {
typedef intrusive_ptr<thread_mapped_actor> ptr_type;
void start() {
ptr_type mt{detail::memory::create<thread_mapped_actor>()};
ptr_type mp{detail::memory::create<thread_mapped_actor>()};
auto mt = make_counted<thread_mapped_actor>();
auto mp = make_counted<thread_mapped_actor>();
// launch threads
m_timer_thread = std::thread{&scheduler_helper::timer_loop, mt};
m_printer_thread = std::thread{&scheduler_helper::printer_loop, mp};
......@@ -177,7 +177,7 @@ void scheduler_helper::timer_loop(scheduler_helper::ptr_type m_self) {
done = true;
},
others() >> [&]() {
# ifdef CPPA_DEBUG
# ifdef CPPA_DEBUG_MODE
std::cerr << "scheduler_helper::timer_loop: UNKNOWN MESSAGE: "
<< to_string(msg_ptr->msg)
<< std::endl;
......
......@@ -31,6 +31,7 @@
#include <pthread.h>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
......@@ -45,6 +46,7 @@ pthread_once_t s_key_once = PTHREAD_ONCE_INIT;
local_actor* tss_constructor() {
local_actor* result = detail::memory::create<thread_mapped_actor>();
CPPA_LOGF_INFO("converted thread to actor; ID = " << result->id());
result->ref();
get_scheduler()->register_converted_context(result);
return result;
......@@ -101,7 +103,7 @@ void self_type::cleanup_fun(cppa::local_actor* what) {
auto ptr = dynamic_cast<thread_mapped_actor*>(what);
if (ptr) {
// make sure "unspawned" actors quit properly
ptr->cleanup(cppa::exit_reason::normal);
what->cleanup(cppa::exit_reason::normal);
}
what->deref();
}
......
......@@ -44,7 +44,6 @@
#include "cppa/detail/group_manager.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/decorated_names_map.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
......@@ -66,7 +65,6 @@ namespace {
std::atomic<opencl::command_dispatcher*> s_command_dispatcher;
std::atomic<uniform_type_info_map*> s_uniform_type_info_map;
std::atomic<decorated_names_map*> s_decorated_names_map;
std::atomic<network::middleman*> s_middleman;
std::atomic<actor_registry*> s_actor_registry;
std::atomic<group_manager*> s_group_manager;
......@@ -77,32 +75,29 @@ std::atomic<logging*> s_logger;
} // namespace <anonymous>
void singleton_manager::shutdown() {
CPPA_LOGF_INFO("prepare to shutdown");
CPPA_LOGF(CPPA_DEBUG, nullptr, "prepare to shutdown");
if (self.unchecked() != nullptr) {
try { self.unchecked()->quit(exit_reason::normal); }
catch (actor_exited&) { }
}
//auto rptr = s_actor_registry.load();
//if (rptr) rptr->await_running_count_equal(0);
CPPA_LOGF_DEBUG("shutdown scheduler");
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown scheduler");
destroy(s_scheduler);
CPPA_LOGF_DEBUG("shutdown middleman");
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown middleman");
destroy(s_middleman);
std::atomic_thread_fence(std::memory_order_seq_cst);
// it's safe to delete all other singletons now
CPPA_LOGF_DEBUG("close OpenCL command dispather");
CPPA_LOGF(CPPA_DEBUG, nullptr, "close OpenCL command dispather");
destroy(s_command_dispatcher);
CPPA_LOGF_DEBUG("close actor registry");
CPPA_LOGF(CPPA_DEBUG, nullptr, "close actor registry");
destroy(s_actor_registry);
CPPA_LOGF_DEBUG("shutdown group manager");
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown group manager");
destroy(s_group_manager);
CPPA_LOGF_DEBUG("destroy empty tuple singleton");
CPPA_LOGF(CPPA_DEBUG, nullptr, "destroy empty tuple singleton");
destroy(s_empty_tuple);
CPPA_LOGF_DEBUG("clear type info map");
CPPA_LOGF(CPPA_DEBUG, nullptr, "clear type info map");
destroy(s_uniform_type_info_map);
CPPA_LOGF_DEBUG("clear decorated names log");
destroy(s_decorated_names_map);
CPPA_LOGF_DEBUG("shutdown logger");
destroy(s_logger);
}
......@@ -131,10 +126,6 @@ scheduler* singleton_manager::get_scheduler() {
return lazy_get(s_scheduler);
}
decorated_names_map* singleton_manager::get_decorated_names_map() {
return lazy_get(s_decorated_names_map);
}
logging* singleton_manager::get_logger() {
return lazy_get(s_logger);
}
......
......@@ -53,7 +53,7 @@ namespace cppa {
namespace {
bool isbuiltin(const string& type_name) {
return type_name == "@str" || type_name == "@atom" || type_name == "@<>";
return type_name == "@str" || type_name == "@atom" || type_name == "@tuple";
}
// serializes types as type_name(...) except:
......@@ -284,7 +284,7 @@ class string_deserializer : public deserializer {
return "@atom";
}
else if (*m_pos == '{') {
return "@<>";
return "@tuple";
}
// default case
auto substr_end = next_delimiter();
......
......@@ -43,43 +43,15 @@
namespace cppa {
thread_mapped_actor::thread_mapped_actor() : super(std::function<void()>{}), m_initialized(true) { }
thread_mapped_actor::thread_mapped_actor() : m_initialized(true) { }
thread_mapped_actor::thread_mapped_actor(std::function<void()> fun)
: super(std::move(fun)), m_initialized(false) { }
/*
auto thread_mapped_actor::init_timeout(const util::duration& rel_time) -> timeout_type {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
return result;
}
void thread_mapped_actor::enqueue(const actor_ptr& sender, any_tuple msg) {
auto ptr = new_mailbox_element(sender, std::move(msg));
this->m_mailbox.push_back(ptr);
}
void thread_mapped_actor::sync_enqueue(const actor_ptr& sender,
message_id id,
any_tuple msg) {
auto ptr = this->new_mailbox_element(sender, std::move(msg), id);
if (!this->m_mailbox.push_back(ptr)) {
detail::sync_request_bouncer f{this->exit_reason()};
f(sender, id);
}
: m_initialized(false) {
set_behavior(std::move(fun));
}
*/
bool thread_mapped_actor::initialized() const {
return m_initialized;
}
void thread_mapped_actor::cleanup(std::uint32_t reason) {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
super::cleanup(reason);
}
} // namespace cppa::detail
......@@ -36,6 +36,7 @@
#include "cppa/on.hpp"
#include "cppa/logging.hpp"
#include "cppa/prioritizing.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/context_switching_actor.hpp"
......@@ -99,24 +100,24 @@ struct thread_pool_scheduler::worker {
actor_ptr next;
for (;;) {
aggressive(job) || moderate(job) || relaxed(job);
CPPA_LOG_DEBUG("dequeued new job");
CPPA_LOGMF(CPPA_DEBUG, self, "dequeued new job");
if (job == m_dummy) {
CPPA_LOG_DEBUG("received dummy (quit)");
CPPA_LOGMF(CPPA_DEBUG, self, "received dummy (quit)");
// dummy of doom received ...
m_job_queue->push_back(job); // kill the next guy
return; // and say goodbye
}
do {
CPPA_LOG_DEBUG("resume actor with ID " << job->id());
CPPA_LOGMF(CPPA_DEBUG, self, "resume actor with ID " << job->id());
CPPA_REQUIRE(next == nullptr);
if (job->resume(&fself, next) == resume_result::actor_done) {
CPPA_LOG_DEBUG("actor is done");
CPPA_LOGMF(CPPA_DEBUG, self, "actor is done");
bool hidden = job->is_hidden();
job->deref();
if (!hidden) get_actor_registry()->dec_running();
}
if (next) {
CPPA_LOG_DEBUG("got new job trough chaining");
CPPA_LOGMF(CPPA_DEBUG, self, "got new job trough chaining");
job = static_cast<job_ptr>(next.get());
next.reset();
}
......@@ -163,11 +164,11 @@ void thread_pool_scheduler::initialize() {
void thread_pool_scheduler::destroy() {
CPPA_LOG_TRACE("");
m_queue.push_back(&m_dummy);
CPPA_LOG_DEBUG("join supervisor");
CPPA_LOGMF(CPPA_DEBUG, self, "join supervisor");
m_supervisor.join();
// make sure job queue is empty, because destructor of m_queue would
// otherwise delete elements it shouldn't
CPPA_LOG_DEBUG("flush queue");
CPPA_LOGMF(CPPA_DEBUG, self, "flush queue");
auto ptr = m_queue.try_pop();
while (ptr != nullptr) {
if (ptr != &m_dummy) {
......@@ -199,7 +200,7 @@ void exec_as_thread(bool is_hidden, local_actor_ptr p, F f) {
}).detach();
}
actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
local_actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
CPPA_REQUIRE(p != nullptr);
bool is_hidden = has_hide_flag(os);
if (has_detach_flag(os)) {
......@@ -218,27 +219,62 @@ actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
return std::move(p);
}
actor_ptr thread_pool_scheduler::exec(spawn_options os,
local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
init_callback cb,
void_function f) {
if (has_blocking_api_flag(os)) {
local_actor_ptr result;
auto set_result = [&](local_actor_ptr value) {
CPPA_REQUIRE(result == nullptr && value != nullptr);
result = std::move(value);
if (cb) cb(result.get());
};
if (has_priority_aware_flag(os)) {
using impl = extend<thread_mapped_actor>::with<prioritizing>;
set_result(make_counted<impl>());
exec_as_thread(has_hide_flag(os), result, [result, f] {
try {
f();
result->exec_behavior_stack();
}
catch (actor_exited& e) { }
catch (std::exception& e) {
CPPA_LOGF_ERROR("actor with ID " << result->id()
<< " terminated due to an unhandled exception; "
<< detail::demangle(typeid(e)) << ": "
<< e.what());
}
catch (...) {
CPPA_LOGF_ERROR("actor with ID " << result->id()
<< " terminated due to an unknown exception");
}
result->on_exit();
});
}
else if (has_blocking_api_flag(os)) {
# ifndef CPPA_DISABLE_CONTEXT_SWITCHING
if (!has_detach_flag(os)) {
return exec(os,
memory::create<context_switching_actor>(std::move(f)));
auto p = make_counted<context_switching_actor>(std::move(f));
set_result(p);
exec(os, std::move(p));
}
else
# endif
thread_mapped_actor_ptr p;
p.reset(memory::create<thread_mapped_actor>(std::move(f)));
/* else tree */ {
auto p = make_counted<thread_mapped_actor>(std::move(f));
set_result(p);
exec_as_thread(has_hide_flag(os), p, [p] {
p->run();
p->on_exit();
});
return std::move(p);
}
}
else {
auto p = event_based_actor::from(std::move(f));
if (cb) cb(p.get());
return exec(os, p);
set_result(p);
exec(os, p);
}
CPPA_REQUIRE(result != nullptr);
return std::move(result);
}
} } // namespace cppa::detail
......@@ -32,6 +32,7 @@
#include <cwchar>
#include <limits>
#include <vector>
#include <cstring>
#include <typeinfo>
#include <stdexcept>
#include <algorithm>
......@@ -42,14 +43,14 @@
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_header.hpp"
#include "cppa/util/void_type.hpp"
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/message_header.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/decorated_names_map.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
namespace {
......@@ -57,6 +58,47 @@ using namespace std;
using namespace cppa;
using namespace detail;
struct platform_int_mapping { const char* name; size_t size; bool is_signed; };
// WARNING: this list is sorted and searched with std::lower_bound;
// keep ordered when adding elements!
platform_int_mapping platform_dependent_sizes[] = {
{"char", sizeof(char), true},
{"char16_t", sizeof(char16_t), true},
{"char32_t", sizeof(char32_t), true},
{"int", sizeof(int), true},
{"long", sizeof(long), true},
{"long int", sizeof(long int), true},
{"long long", sizeof(long long), true},
{"short", sizeof(short), true},
{"short int", sizeof(short int), true},
{"signed char", sizeof(signed char), true},
{"signed int", sizeof(signed int), true},
{"signed long", sizeof(signed long), true},
{"signed long int", sizeof(signed long int), true},
{"signed long long", sizeof(signed long long), true},
{"signed short", sizeof(signed short), true},
{"signed short int", sizeof(signed short int), true},
{"unsigned char", sizeof(unsigned char), false},
{"unsigned int", sizeof(unsigned int), false},
{"unsigned long", sizeof(unsigned long), false},
{"unsigned long int", sizeof(unsigned long int), false},
{"unsigned long long", sizeof(unsigned long long), false},
{"unsigned short", sizeof(unsigned short), false},
{"unsigned short int", sizeof(unsigned short int), false}
};
string map2decorated(const char* name) {
auto cmp = [](const platform_int_mapping& pim, const char* name) {
return strcmp(pim.name, name) < 0;
};
auto e = end(platform_dependent_sizes);
auto i = lower_bound(begin(platform_dependent_sizes), e, name, cmp);
if (i != e && strcmp(i->name, name) == 0) {
return mapped_int_names[i->size][i->is_signed ? 1 : 0];
}
return mapped_name_by_decorated_name(name);
}
class parse_tree {
......@@ -67,7 +109,7 @@ class parse_tree {
if (m_volatile) result += "volatile ";
if (m_const) result += "const ";
if (!m_template) {
result += dmm->decorate(m_name);
result += map2decorated(m_name.c_str());
}
else {
string full_name = m_name;
......@@ -79,7 +121,7 @@ class parse_tree {
}
full_name += ">";
// decorate full name
result += dmm->decorate(full_name);
result += map2decorated(full_name.c_str());
}
if (m_pointer) result += "*";
if (m_lvalue_ref) result += "&";
......@@ -167,9 +209,7 @@ class parse_tree {
parse_tree()
: m_const(false), m_pointer(false), m_volatile(false), m_template(false)
, m_lvalue_ref(false), m_rvalue_ref(false) {
dmm = singleton_manager::get_decorated_names_map();
}
, m_lvalue_ref(false), m_rvalue_ref(false) { }
bool m_const;
bool m_pointer;
......@@ -177,7 +217,6 @@ class parse_tree {
bool m_template;
bool m_lvalue_ref;
bool m_rvalue_ref;
const decorated_names_map* dmm;
string m_name;
vector<parse_tree> m_template_parameters;
......@@ -221,7 +260,7 @@ void replace_all(string& str, const char (&before)[RawSize], const char* after)
}
const char s_rawan[] = "anonymous namespace";
const char s_an[] = "@_";
const char s_an[] = "$";
} // namespace <anonymous>
......
This diff is collapsed.
This diff is collapsed.
......@@ -16,12 +16,14 @@ namespace {
size_t s_pongs = 0;
behavior ping_behavior(size_t num_pings) {
CPPA_LOGF_TRACE("");
return (
on(atom("pong"), arg_match) >> [num_pings](int value) {
CPPA_LOGF_ERROR_IF(!self->last_sender(), "last_sender() == nullptr");
CPPA_LOGF_INFO("received {'pong', " << value << "}");
//cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs >= num_pings) {
CPPA_LOGF_INFO("reached maximum, send {'EXIT', user_defined} "
<< "to last sender and quit with normal reason");
send_exit(self->last_sender(), exit_reason::user_defined);
self->quit();
}
......@@ -30,25 +32,20 @@ behavior ping_behavior(size_t num_pings) {
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
<< to_string(self->last_dequeued()));
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
self->quit(exit_reason::user_defined);
}
);
}
behavior pong_behavior() {
CPPA_LOGF_TRACE("");
return (
on<atom("ping"), int>() >> [](int value) {
//cout << to_string(self->last_dequeued()) << endl;
on(atom("ping"), arg_match) >> [](int value) {
CPPA_LOGF_INFO("received {'ping', " << value << "}");
reply(atom("pong"), value + 1);
},
others() >> []() {
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
<< to_string(self->last_dequeued()));
self->quit(exit_reason::user_defined);
}
);
......@@ -61,31 +58,30 @@ size_t pongs() {
}
void ping(size_t num_pings) {
CPPA_SET_DEBUG_NAME("ping");
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
receive_loop(ping_behavior(num_pings));
}
actor_ptr spawn_event_based_ping(size_t num_pings) {
void event_based_ping(size_t num_pings) {
CPPA_SET_DEBUG_NAME("event_based_ping");
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
return spawn([=] {
become(ping_behavior(num_pings));
});
}
void pong(actor_ptr ping_actor) {
CPPA_SET_DEBUG_NAME("pong");
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
// kickoff
send(ping_actor, atom("pong"), 0);
receive_loop (pong_behavior());
send(ping_actor, atom("pong"), 0); // kickoff
receive_loop(pong_behavior());
}
actor_ptr spawn_event_based_pong(actor_ptr ping_actor) {
void event_based_pong(actor_ptr ping_actor) {
CPPA_SET_DEBUG_NAME("event_based_pong");
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
CPPA_REQUIRE(ping_actor.get() != nullptr);
return factory::event_based([=] {
self->become(pong_behavior());
send(ping_actor, atom("pong"), 0);
}).spawn();
CPPA_REQUIRE(ping_actor != nullptr);
send(ping_actor, atom("pong"), 0); // kickoff
become(pong_behavior());
}
......@@ -8,11 +8,11 @@
void ping(size_t num_pings);
cppa::actor_ptr spawn_event_based_ping(size_t num_pings);
void event_based_ping(size_t num_pings);
void pong(cppa::actor_ptr ping_actor);
cppa::actor_ptr spawn_event_based_pong(cppa::actor_ptr ping_actor);
void event_based_pong(cppa::actor_ptr ping_actor);
// returns the number of messages ping received
size_t pongs();
......
......@@ -3,6 +3,7 @@
#include <vector>
#include <string>
#include <cstring>
#include <cstddef>
#include <sstream>
#include <iostream>
......@@ -29,9 +30,9 @@ void cppa_unexpected_timeout(const char* fname, size_t line_num);
#define CPPA_PRINT(message) CPPA_PRINTC(__FILE__, __LINE__, message)
#define CPPA_PRINTERRC(filename, linenum, message) \
CPPA_LOGF_ERROR(CPPA_STREAMIFY(filename, linenum, message)); \
std::cerr << "ERROR: " << CPPA_STREAMIFY(filename, linenum, message) \
#define CPPA_PRINTERRC(fname, linenum, msg) \
CPPA_LOGF(CPPA_ERROR, ::cppa::self, CPPA_STREAMIFY(fname, linenum, msg)); \
std::cerr << "ERROR: " << CPPA_STREAMIFY(fname, linenum, msg) \
<< std::endl
#define CPPA_PRINTERR(message) CPPA_PRINTERRC(__FILE__, __LINE__, message)
......@@ -43,7 +44,9 @@ struct both_integral {
};
template<bool V, typename T1, typename T2>
struct enable_integral : std::enable_if<both_integral<T1,T2>::value == V> { };
struct enable_integral : std::enable_if< both_integral<T1,T2>::value == V
&& not std::is_pointer<T1>::value
&& not std::is_pointer<T2>::value> { };
template<typename T>
const T& cppa_stream_arg(const T& value) {
......@@ -73,6 +76,15 @@ inline void cppa_failed(const V1& v1,
cppa_inc_error_count();
}
inline void cppa_check_value(const std::string& v1,
const std::string& v2,
const char* fname,
int line,
bool expected = true) {
if ((v1 == v2) == expected) cppa_passed(fname, line);
else cppa_failed(v1, v2, fname, line);
}
template<typename V1, typename V2>
inline void cppa_check_value(const V1& v1,
const V2& v2,
......@@ -98,10 +110,11 @@ inline void cppa_check_value(V1 v1,
#define CPPA_VERBOSE_EVAL(LineOfCode) \
CPPA_PRINT(#LineOfCode << " = " << (LineOfCode));
#define CPPA_TEST(name) \
#define CPPA_TEST(testname) \
auto cppa_test_scope_guard = ::cppa::util::make_scope_guard([] { \
std::cout << cppa_error_count() << " error(s) detected" << std::endl; \
});
}); \
CPPA_LOGF_INFO("run unit test " << #testname)
#define CPPA_TEST_RESULT() ((cppa_error_count() == 0) ? 0 : -1)
......@@ -119,7 +132,7 @@ inline void cppa_check_value(V1 v1,
CPPA_PRINTERR(#line_of_code); \
cppa_inc_error_count(); \
} \
else CPPA_PRINT("passed")
else { CPPA_PRINT("passed"); } CPPA_VOID_STMT
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) \
cppa_check_value((lhs_loc), (rhs_loc), __FILE__, __LINE__)
......@@ -127,7 +140,7 @@ inline void cppa_check_value(V1 v1,
#define CPPA_CHECK_NOT_EQUAL(rhs_loc, lhs_loc) \
cppa_check_value((lhs_loc), (rhs_loc), __FILE__, __LINE__, false)
#define CPPA_ERROR(err_msg) { \
#define CPPA_FAILURE(err_msg) { \
CPPA_PRINTERR("ERROR: " << err_msg); \
cppa_inc_error_count(); \
} ((void) 0)
......@@ -143,7 +156,7 @@ inline void cppa_check_value(V1 v1,
// some convenience macros for defining callbacks
#define CPPA_CHECKPOINT_CB() [] { CPPA_CHECKPOINT(); }
#define CPPA_ERROR_CB(err_msg) [] { CPPA_ERROR(err_msg); }
#define CPPA_FAILURE_CB(err_msg) [] { CPPA_FAILURE(err_msg); }
#define CPPA_UNEXPECTED_MSG_CB() [] { CPPA_UNEXPECTED_MSG(); }
#define CPPA_UNEXPECTED_TOUT_CB() [] { CPPA_UNEXPECTED_TOUT(); }
......
......@@ -177,7 +177,7 @@ int main() {
invoked = true;
}
);
if (!invoked) { CPPA_ERROR("match(\"abc\") failed"); }
if (!invoked) { CPPA_FAILURE("match(\"abc\") failed"); }
invoked = false;
bool disable_case1 = true;
......@@ -223,7 +223,7 @@ int main() {
invoked = true;
}
);
if (!invoked) { CPPA_ERROR("match({1, 2, 3}) failed"); }
if (!invoked) { CPPA_FAILURE("match({1, 2, 3}) failed"); }
invoked = false;
string sum;
......@@ -240,17 +240,11 @@ int main() {
on<char>().when(_x1.in({'w', 't', 'f'})) >> [&](char c) {
sum += c;
},
on<char>() >> [&](char c) {
CPPA_ERROR("whaaaaat? guard didn't match: " << c);
},
others() >> [&]() {
CPPA_ERROR("unexpected match");
}
on<char>() >> CPPA_FAILURE_CB("guard didn't match"),
others() >> CPPA_FAILURE_CB("unexpected match")
);
},
others() >> [&]() {
CPPA_ERROR("unexpected match");
}
others() >> CPPA_FAILURE_CB("unexpected match")
);
CPPA_CHECK_EQUAL(sum, "-h--versionwtf");
......@@ -270,7 +264,7 @@ int main() {
str = "C";
}
);
if (!invoked) { CPPA_ERROR("match({\"a\", \"b\", \"c\"}) failed"); }
if (!invoked) { CPPA_FAILURE("match({\"a\", \"b\", \"c\"}) failed"); }
CPPA_CHECK_EQUAL(vec.back(), "C");
invoked = false;
......@@ -280,7 +274,7 @@ int main() {
str = "A";
}
);
if (!invoked) { CPPA_ERROR("match_each({\"a\", \"b\", \"C\"}) failed"); }
if (!invoked) { CPPA_FAILURE("match_each({\"a\", \"b\", \"C\"}) failed"); }
CPPA_CHECK_EQUAL(vec.front(), "A");
invoked = false;
......@@ -328,7 +322,7 @@ int main() {
else return false;
},
on_arg_match >> [](const string& arg) -> bool{
CPPA_ERROR("unexpected string: " << arg);
CPPA_FAILURE("unexpected string: " << arg);
return false;
}
);
......
......@@ -137,7 +137,7 @@ int main() {
auto map_args = [] (any_tuple msg) -> option<cow_tuple<ivec>> {
auto opt = tuple_cast<matrix_type>(msg);
if (opt) {
return {move(get_ref<0>(*opt).data())};
return {make_cow_tuple(move(get_ref<0>(*opt).data()))};
}
return {};
};
......
This diff is collapsed.
......@@ -165,7 +165,7 @@ int main() {
CPPA_CHECK_EQUAL(get<1>(tup), "foo");
}
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
catch (exception& e) { CPPA_FAILURE(to_verbose_string(e)); }
try {
// test raw_type in both binary and string serialization
......@@ -184,7 +184,7 @@ int main() {
rs2 = from_string<raw_struct>(rsstr);
CPPA_CHECK_EQUAL(rs2.str, rs.str);
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
catch (exception& e) { CPPA_FAILURE(to_verbose_string(e)); }
try {
auto ttup = make_any_tuple(1, 2, actor_ptr(self));
......@@ -201,7 +201,7 @@ int main() {
CPPA_CHECK(ttup == ttup3);
CPPA_CHECK(ttup2 == ttup3);
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
catch (exception& e) { CPPA_FAILURE(to_verbose_string(e)); }
try {
// serialize b1 to buf
......@@ -221,14 +221,15 @@ int main() {
CPPA_CHECK_EQUAL(get<1>(tup), "foo");
}
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
catch (exception& e) { CPPA_FAILURE(to_verbose_string(e)); }
try {
any_tuple msg1 = cppa::make_cow_tuple(42, string("Hello \"World\"!"));
auto msg1_tostring = to_string(msg1);
if (msg1str != msg1_tostring) {
CPPA_ERROR("msg1str != to_string(msg1)");
CPPA_FAILURE("msg1str != to_string(msg1)");
cerr << "to_string(msg1) = " << msg1_tostring << endl;
cerr << "to_string(msg1str) = " << msg1str << endl;
}
util::buffer wr_buf;
binary_serializer bs(&wr_buf, &addressing);
......@@ -256,10 +257,10 @@ int main() {
}
}
else {
CPPA_ERROR("obj.type() != typeid(message)");
CPPA_FAILURE("obj.type() != typeid(message)");
}
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
catch (exception& e) { CPPA_FAILURE(to_verbose_string(e)); }
CPPA_CHECK((is_iterable<int>::value) == false);
// string is primitive and thus not identified by is_iterable
......
......@@ -282,6 +282,35 @@ struct simple_mirror : sb_actor<simple_mirror> {
};
void high_priority_testee() {
send(self, atom("b"));
send({self, message_priority::high}, atom("a"));
// 'a' must be received before 'b'
become (
on(atom("b")) >> [] {
CPPA_FAILURE("received 'b' before 'a'");
self->quit();
},
on(atom("a")) >> [] {
CPPA_CHECKPOINT();
become (
on(atom("b")) >> [] {
CPPA_CHECKPOINT();
self->quit();
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
struct high_priority_testee_class : event_based_actor {
void init() {
high_priority_testee();
}
};
int main() {
CPPA_TEST(test_spawn);
......@@ -334,6 +363,23 @@ int main() {
CPPA_CHECKPOINT();
}
CPPA_PRINT("test priority aware mirror"); {
auto mirror = spawn<simple_mirror,monitored+priority_aware>();
CPPA_CHECKPOINT();
send(mirror, "hello mirror");
receive (
on("hello mirror") >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
send_exit(mirror, exit_reason::user_defined);
receive (
on(atom("DOWN"), exit_reason::user_defined) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_others_done();
CPPA_CHECKPOINT();
}
CPPA_PRINT("test echo actor");
auto mecho = spawn(echo_actor);
send(mecho, "hello echo");
......@@ -373,7 +419,7 @@ int main() {
CPPA_CHECKPOINT();
auto factory = factory::event_based([&](int* i, float*, string*) {
self->become (
become (
on(atom("get_int")) >> [i]() {
reply(*i);
},
......@@ -474,7 +520,7 @@ int main() {
CPPA_PRINT("test sync send with factory spawned actor");
auto sync_testee_factory = factory::event_based(
[&]() {
self->become (
become (
on("hi") >> [&]() {
auto handle = sync_send(self->last_sender(), "whassup?");
handle_response(handle) (
......@@ -526,7 +572,7 @@ int main() {
auto inflater = factory::event_based(
[](string*, actor_ptr* receiver) {
self->become(
become(
on_arg_match >> [=](int n, const string& s) {
send(*receiver, n * 2, s);
},
......@@ -560,7 +606,7 @@ int main() {
if (*name == "Joe" && !*pal) {
*pal = spawn_next("Bob", self);
}
self->become (
become (
others() >> [pal]() {
// forward message and die
*pal << self->last_dequeued();
......@@ -605,7 +651,7 @@ int main() {
CPPA_CHECK_EQUAL(zombie_on_exit_called, 3);
auto f = factory::event_based([](string* name) {
self->become (
become (
on(atom("get_name")) >> [name]() {
reply(atom("name"), *name);
}
......@@ -630,7 +676,7 @@ int main() {
await_all_others_done();
factory::event_based([](int* i) {
self->become(
become(
after(chrono::milliseconds(50)) >> [=]() {
if (++(*i) >= 5) self->quit();
}
......@@ -685,14 +731,23 @@ int main() {
flags |= 0x08;
},
others() >> [&]() {
CPPA_ERROR("unexpected message: " << to_string(self->last_dequeued()));
CPPA_FAILURE("unexpected message: " << to_string(self->last_dequeued()));
},
after(chrono::seconds(5)) >> [&]() {
CPPA_ERROR("timeout in file " << __FILE__ << " in line " << __LINE__);
CPPA_FAILURE("timeout in file " << __FILE__ << " in line " << __LINE__);
}
);
// wait for termination of all spawned actors
await_all_others_done();
CPPA_CHECK_EQUAL(flags, 0x0F);
// verify pong messages
CPPA_CHECK_EQUAL(pongs(), 10);
CPPA_CHECKPOINT();
spawn<priority_aware>(high_priority_testee);
await_all_others_done();
CPPA_CHECKPOINT();
spawn<high_priority_testee_class,priority_aware>();
await_all_others_done();
// don't try this at home, kids
send(self, atom("check"));
try {
......@@ -703,14 +758,11 @@ int main() {
}
);
self->exec_behavior_stack();
CPPA_ERROR("line " << __LINE__ << " should be unreachable");
CPPA_FAILURE("line " << __LINE__ << " should be unreachable");
}
catch (actor_exited&) {
CPPA_CHECKPOINT();
}
CPPA_CHECK_EQUAL(flags, 0x0F);
// verify pong messages
CPPA_CHECK_EQUAL(pongs(), 10);
shutdown();
return CPPA_TEST_RESULT();
}
......@@ -49,7 +49,9 @@ struct A : popular_actor {
void init() {
become (
on(atom("go"), arg_match) >> [=](const actor_ptr& next) {
CPPA_CHECKPOINT();
sync_send(next, atom("gogo")).then([=] {
CPPA_CHECKPOINT();
send(buddy(), atom("success"));
quit();
});
......@@ -59,29 +61,18 @@ struct A : popular_actor {
}
};
#ifdef __clang__
struct B : sb_actor<B,popular_actor> {
B(const actor_ptr& buddy) : sb_actor<B,popular_actor>(buddy) { }
behavior init_state = (
others() >> [=] {
forward_to(buddy());
quit();
}
);
};
#else
struct B : popular_actor {
B(const actor_ptr& buddy) : popular_actor(buddy) { }
void init() {
become (
others() >> [=] {
CPPA_CHECKPOINT();
forward_to(buddy());
quit();
}
);
}
};
#endif
struct C : sb_actor<C> {
behavior init_state = (
......@@ -168,7 +159,7 @@ struct server : event_based_actor {
int main() {
CPPA_TEST(test_sync_send);
self->on_sync_failure([] {
CPPA_ERROR("received: " << to_string(self->last_dequeued()));
CPPA_FAILURE("received: " << to_string(self->last_dequeued()));
});
spawn<monitored + blocking_api>([] {
CPPA_LOGC_TRACE("NONE", "main$sync_failure_test", "id = " << self->id());
......@@ -177,7 +168,7 @@ int main() {
send(foi, atom("i"));
receive(on_arg_match >> [](int i) { CPPA_CHECK_EQUAL(i, 0); });
self->on_sync_failure([] {
CPPA_ERROR("received: " << to_string(self->last_dequeued()));
CPPA_FAILURE("received: " << to_string(self->last_dequeued()));
});
sync_send(foi, atom("i")).then(
[&](int i) { CPPA_CHECK_EQUAL(i, 0); ++invocations; },
......@@ -191,6 +182,7 @@ int main() {
});
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(invocations, 2);
CPPA_PRINT("trigger sync failure");
// provoke invocation of self->handle_sync_failure()
bool sync_failure_called = false;
bool int_handler_called = false;
......@@ -198,7 +190,9 @@ int main() {
sync_failure_called = true;
});
sync_send(foi, atom("f")).await(
on<int>() >> [&] { int_handler_called = true; }
on<int>() >> [&] {
int_handler_called = true;
}
);
CPPA_CHECK_EQUAL(sync_failure_called, true);
CPPA_CHECK_EQUAL(int_handler_called, false);
......@@ -221,10 +215,10 @@ int main() {
auto await_success_message = [&] {
receive (
on(atom("success")) >> CPPA_CHECKPOINT_CB(),
on(atom("failure")) >> CPPA_ERROR_CB("A didn't receive sync response"),
on(atom("failure")) >> CPPA_FAILURE_CB("A didn't receive sync response"),
on(atom("DOWN"), arg_match).when(_x2 != exit_reason::normal)
>> [&](uint32_t err) {
CPPA_ERROR("A exited for reason " << err);
CPPA_FAILURE("A exited for reason " << err);
}
);
};
......@@ -263,8 +257,8 @@ int main() {
self->on_sync_timeout([&] { timeout_occured = true; });
self->on_sync_failure(CPPA_UNEXPECTED_MSG_CB());
timed_sync_send(c, std::chrono::milliseconds(500), atom("HiThere"))
.then(CPPA_ERROR_CB("C replied to 'HiThere'!"))
.continue_with(CPPA_ERROR_CB("bad continuation"));
.then(CPPA_FAILURE_CB("C replied to 'HiThere'!"))
.continue_with(CPPA_FAILURE_CB("bad continuation"));
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(timeout_occured, true);
self->on_sync_failure(CPPA_UNEXPECTED_MSG_CB());
......
......@@ -79,12 +79,12 @@ option<int> str2int(const std::string& str) {
#define CPPA_CHECK_INVOKED(FunName, Args) \
if ( ( FunName Args ) == false || invoked != #FunName ) { \
CPPA_ERROR("invocation of " #FunName " failed"); \
CPPA_FAILURE("invocation of " #FunName " failed"); \
} invoked = ""
#define CPPA_CHECK_NOT_INVOKED(FunName, Args) \
if ( ( FunName Args ) == true || invoked == #FunName ) { \
CPPA_ERROR(#FunName " erroneously invoked"); \
CPPA_FAILURE(#FunName " erroneously invoked"); \
} invoked = ""
struct dummy_receiver : event_based_actor {
......@@ -222,13 +222,13 @@ void check_guards() {
CPPA_CHECK_EQUAL(f09_any_val.get_as<int>(1), 9);
f09_any_val.get_as_mutable<int>(1) = 666;
any_tuple f09_any_val_copy{f09_any_val};
CPPA_CHECK_EQUAL(f09_any_val.at(0), f09_any_val_copy.at(0));
CPPA_CHECK(f09_any_val.at(0) == f09_any_val_copy.at(0));
// detaches f09_any_val from f09_any_val_copy
CPPA_CHECK(f09.invoke(f09_any_val));
CPPA_CHECK_EQUAL(f09_any_val.get_as<int>(1), 9);
CPPA_CHECK_EQUAL(f09_any_val_copy.get_as<int>(1), 666);
// no longer the same data
CPPA_CHECK_NOT_EQUAL(f09_any_val.at(0), f09_any_val_copy.at(0));
CPPA_CHECK(f09_any_val.at(0) != f09_any_val_copy.at(0));
auto f10 = (
on<int>().when(_x1 < 10) >> [&]() { invoked = "f10.0"; },
......@@ -351,11 +351,11 @@ void check_wildcards() {
CPPA_CHECK_EQUAL(get<0>(v0), "1");
CPPA_CHECK_EQUAL(get<0>(t0), get<0>(v0));
// check cow semantics
CPPA_CHECK_EQUAL(&get<0>(t0), &get<0>(v0)); // point to the same
CPPA_CHECK(&get<0>(t0) == &get<0>(v0)); // point to the same
get_ref<0>(t0) = "hello world"; // detaches t0 from v0
CPPA_CHECK_EQUAL(get<0>(t0), "hello world"); // t0 contains new value
CPPA_CHECK_EQUAL(get<0>(v0), "1"); // v0 contains old value
CPPA_CHECK_NOT_EQUAL(&get<0>(t0), &get<0>(v0)); // no longer the same
CPPA_CHECK(&get<0>(t0) != &get<0>(v0)); // no longer the same
// check operator==
auto lhs = make_cow_tuple(1,2,3,4);
auto rhs = make_cow_tuple(static_cast<std::uint8_t>(1), 2.0, 3, 4);
......@@ -368,24 +368,24 @@ void check_wildcards() {
CPPA_CHECK(opt0);
if (opt0) {
CPPA_CHECK((*opt0 == make_cow_tuple("one", 2, 3.f, 4.0)));
CPPA_CHECK_EQUAL(&get<0>(*opt0), at1.at(0));
CPPA_CHECK_EQUAL(&get<1>(*opt0), at1.at(1));
CPPA_CHECK_EQUAL(&get<2>(*opt0), at1.at(2));
CPPA_CHECK_EQUAL(&get<3>(*opt0), at1.at(3));
CPPA_CHECK(&get<0>(*opt0) == at1.at(0));
CPPA_CHECK(&get<1>(*opt0) == at1.at(1));
CPPA_CHECK(&get<2>(*opt0) == at1.at(2));
CPPA_CHECK(&get<3>(*opt0) == at1.at(3));
}
// leading wildcard
auto opt1 = tuple_cast<anything, double>(at1);
CPPA_CHECK(opt1);
if (opt1) {
CPPA_CHECK_EQUAL(get<0>(*opt1), 4.0);
CPPA_CHECK_EQUAL(&get<0>(*opt1), at1.at(3));
CPPA_CHECK(&get<0>(*opt1) == at1.at(3));
}
// trailing wildcard
auto opt2 = tuple_cast<std::string, anything>(at1);
CPPA_CHECK(opt2);
if (opt2) {
CPPA_CHECK_EQUAL(get<0>(*opt2), "one");
CPPA_CHECK_EQUAL(&get<0>(*opt2), at1.at(0));
CPPA_CHECK(&get<0>(*opt2) == at1.at(0));
}
// wildcard in between
auto opt3 = tuple_cast<std::string, anything, double>(at1);
......@@ -394,8 +394,8 @@ void check_wildcards() {
CPPA_CHECK((*opt3 == make_cow_tuple("one", 4.0)));
CPPA_CHECK_EQUAL(get<0>(*opt3), "one");
CPPA_CHECK_EQUAL(get<1>(*opt3), 4.0);
CPPA_CHECK_EQUAL(&get<0>(*opt3), at1.at(0));
CPPA_CHECK_EQUAL(&get<1>(*opt3), at1.at(3));
CPPA_CHECK(&get<0>(*opt3) == at1.at(0));
CPPA_CHECK(&get<1>(*opt3) == at1.at(3));
}
}
}
......
This diff is collapsed.
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