Commit b6d1575e authored by Joseph Noir's avatar Joseph Noir

Merge branch 'unstable' into topic/opencl

Conflicts:
	cppa.files
parents 922eeb79 55bc8869
......@@ -2,8 +2,8 @@ cmake_minimum_required(VERSION 2.8)
project(cppa CXX)
set(LIBCPPA_VERSION_MAJOR 0)
set(LIBCPPA_VERSION_MINOR 5)
set(LIBCPPA_VERSION_PATCH 5)
set(LIBCPPA_VERSION_MINOR 6)
set(LIBCPPA_VERSION_PATCH 0)
# prohibit in-source builds
if ("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_BINARY_DIR}")
......@@ -116,6 +116,7 @@ set(LIBCPPA_SRC
src/empty_tuple.cpp
src/event_based_actor.cpp
src/exception.cpp
src/exit_reason.cpp
src/factory.cpp
src/fd_util.cpp
src/fiber.cpp
......@@ -132,6 +133,7 @@ set(LIBCPPA_SRC
src/middleman.cpp
src/object.cpp
src/object_array.cpp
src/on.cpp
src/opt.cpp
src/partial_function.cpp
src/primitive_variant.cpp
......
Version 0.6
-----------
__2013_02_22__
- Added `quit_actor` function to terminate actors
- Added continuation feature to non-blocking API
- Allow functor-only usage of `then` and `await`
- Support for Boost 1.53
- Fixed timing bug in synchronous response handling
- Better diagnostics in unit tests
- Use -O3 for release build, because -O4 is broken in Clang
- Auto-reply `EXITED` to orphaned sync requests
- Added `partial_function::or_else` to concatenate partial functions
- New `timed_sync_send` API with different timeout handling
- Added `on_sync_failure` and `on_sync_timeout` handlers for sync messaging
- Added `skip_message` helper to allow users to skip messages manually
Version 0.5.5
-------------
......
......@@ -31,7 +31,7 @@ PROJECT_NAME = libcppa
# This could be handy for archiving the generated documentation or
# if some version control system is used.
PROJECT_NUMBER = "Version 0.5.5"
PROJECT_NUMBER = "Version 0.6.0"
# The OUTPUT_DIRECTORY tag is used to specify the (relative or absolute)
# base path where the generated documentation will be put.
......
......@@ -302,3 +302,5 @@ src/opencl/global.cpp
src/opencl/program.cpp
src/opencl/actor_facade.cpp
cppa/opencl/smart_ptr.hpp
src/exit_reason.cpp
src/on.cpp
......@@ -577,16 +577,6 @@ message_future timed_sync_send(const actor_ptr& whom,
make_any_tuple(std::forward<Args>(what)...));
}
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid
* timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
sync_recv_helper receive_response(const message_future& handle);
/**
* @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple.
......
......@@ -34,6 +34,7 @@
#include <vector>
#include <memory>
#include <utility>
#include <algorithm>
#include "cppa/option.hpp"
#include "cppa/config.hpp"
......@@ -43,70 +44,76 @@
namespace cppa { namespace detail {
class behavior_stack_help_iterator;
struct behavior_stack_mover;
class behavior_stack
{
friend class behavior_stack_help_iterator;
friend struct behavior_stack_mover;
behavior_stack(const behavior_stack&) = delete;
behavior_stack& operator=(const behavior_stack&) = delete;
typedef std::pair<behavior, message_id_t> element_type;
typedef std::pair<behavior,message_id_t> element_type;
public:
behavior_stack() = default;
inline bool empty() const { return m_elements.empty(); }
inline behavior& back() {
CPPA_REQUIRE(!empty());
return m_elements.back().first;
}
// @pre expected_response.valid()
option<behavior&> sync_handler(message_id_t expected_response);
// erases the last asynchronous message handler
void pop_async_back();
// erases the synchronous response handler associated with @p response_id
void erase(message_id_t response_id);
void clear();
void push_back(behavior&& what,
message_id_t expected_response = message_id_t());
// erases the synchronous response handler associated with @p rid
void erase(message_id_t rid) {
erase_if([=](const element_type& e) { return e.second == rid; });
}
void cleanup();
inline bool empty() const { return m_elements.empty(); }
void clear();
inline behavior& back() {
CPPA_REQUIRE(!empty());
return m_elements.back().first;
}
inline void push_back(behavior&& what,
message_id_t response_id = message_id_t::invalid) {
m_elements.emplace_back(std::move(what), response_id);
}
inline void cleanup() {
m_erased_elements.clear();
}
template<class Policy, class Client>
bool invoke(Policy& policy, Client* client, recursive_queue_node* node) {
CPPA_REQUIRE(!m_elements.empty());
CPPA_REQUIRE(!empty());
CPPA_REQUIRE(client != nullptr);
CPPA_REQUIRE(node != nullptr);
// use a copy, because the invoked behavior might change m_elements
behavior what = m_elements.back().first;
// use a copy of bhvr, because invoked behavior might change m_elements
auto id = m_elements.back().second;
if (policy.invoke(client, node, what, id)) {
auto bhvr = m_elements.back().first;
if (policy.invoke(client, node, bhvr, id)) {
bool repeat;
// try to match cached messages
do {
// remove synchronous response handler if needed
if (id.valid()) {
auto last = m_elements.end();
auto i = std::find_if(m_elements.begin(), last,
[id](element_type& e) {
return id == e.second;
});
if (i != last) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
erase_if([id](const element_type& value) {
return id == value.second;
});
}
if (!empty()) {
id = m_elements.back().second;
bhvr = m_elements.back().first;
repeat = policy.invoke_from_cache(client, bhvr, id);
}
id = empty() ? message_id_t() : m_elements.back().second;
} while (!empty() && policy.invoke_from_cache(client, back(), id));
else repeat = false;
} while (repeat);
return true;
}
return false;
......@@ -125,6 +132,29 @@ class behavior_stack
std::vector<element_type> m_elements;
std::vector<behavior> m_erased_elements;
// note: checks wheter i points to m_elements.end() before calling erase()
inline void erase_at(std::vector<element_type>::iterator i) {
if (i != m_elements.end()) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
}
inline void erase_at(std::vector<element_type>::reverse_iterator i) {
// base iterator points to the element *after* the correct element
if (i != m_elements.rend()) erase_at(i.base() - 1);
}
template<typename UnaryPredicate>
inline void erase_if(UnaryPredicate p) {
erase_at(std::find_if(m_elements.begin(), m_elements.end(), p));
}
template<typename UnaryPredicate>
inline void rerase_if(UnaryPredicate p) {
erase_at(std::find_if(m_elements.rbegin(), m_elements.rend(), p));
}
};
} } // namespace cppa::detail
......
......@@ -219,6 +219,7 @@ class receive_policy {
expired_timeout_message,
expired_sync_response,
timeout_message,
timeout_response_message,
ordinary_message,
sync_response
};
......@@ -232,7 +233,7 @@ class receive_policy {
filter_result filter_msg(Client* client, pointer node) {
const any_tuple& msg = node->msg;
auto message_id = node->mid;
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
auto& arr = detail::static_types_array<atom_value,std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.type_at(1) == arr[1]) {
......@@ -255,6 +256,12 @@ class receive_policy {
: expired_timeout_message;
}
}
else if ( msg.size() == 1
&& msg.type_at(0) == arr[0]
&& msg.get_as<atom_value>(0) == atom("TIMEOUT")
&& message_id.is_response()) {
return timeout_response_message;
}
if (message_id.is_response()) {
return (client->awaits(message_id)) ? sync_response
: expired_sync_response;
......@@ -337,6 +344,7 @@ class receive_policy {
Fun& fun,
message_id_t awaited_response,
Policy policy ) {
bool handle_sync_failure_on_mismatch = true;
if (hm_should_skip(node, policy)) {
return hm_skip_msg;
}
......@@ -356,10 +364,16 @@ class receive_policy {
}
return hm_msg_handled;
}
case timeout_response_message: {
handle_sync_failure_on_mismatch = false;
// fall through
}
case sync_response: {
if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = hm_begin(client, node, policy);
fun(node->msg);
if (!fun(node->msg) && handle_sync_failure_on_mismatch) {
client->handle_sync_failure();
}
client->mark_arrived(awaited_response);
client->remove_handler(awaited_response);
hm_cleanup(client, previous_node, policy);
......
......@@ -82,6 +82,8 @@ static constexpr std::uint32_t remote_link_unreachable = 0x00101;
*/
static constexpr std::uint32_t user_defined = 0x10000;
const char* as_string(std::uint32_t value);
} } // namespace cppa::exit_reason
#endif // CPPA_EXIT_REASON_HPP
......@@ -56,10 +56,10 @@ namespace cppa {
class scheduler;
class message_future;
class local_scheduler;
class sync_handle_helper;
namespace detail { class receive_policy; }
template<bool DiscardOld>
struct behavior_policy { static const bool discard_old = DiscardOld; };
......@@ -89,27 +89,6 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t();
} // namespace <anonymous>
#endif // CPPA_DOCUMENTATION
struct sync_recv_helper {
typedef void (*callback_type)(behavior&, message_id_t);
message_id_t m_handle;
callback_type m_fun;
inline sync_recv_helper(message_id_t handle, callback_type fun)
: m_handle(handle), m_fun(fun) { }
template<typename... Expression>
inline void operator()(Expression&&... mexpr) const {
auto bhvr = match_expr_convert(std::forward<Expression>(mexpr)...);
static_assert(std::is_same<decltype(bhvr), behavior>::value,
"no timeout specified");
if (bhvr.timeout().valid() == false || bhvr.timeout().is_zero()) {
throw std::invalid_argument("specified timeout is invalid or zero");
}
else if (!m_handle.valid() || !m_handle.is_response()) {
throw std::logic_error("handle does not point to a response");
}
m_fun(bhvr, m_handle);
}
};
class message_future;
//inline sync_recv_helper receive_response(message_future);
......@@ -307,16 +286,6 @@ class local_actor : public memory_cached_mixin<actor> {
true);
}
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid
* timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
sync_recv_helper handle_response(const message_future& handle);
/**
* @brief Returns to a previous behavior if available.
*/
......@@ -404,6 +373,10 @@ class local_actor : public memory_cached_mixin<actor> {
# ifndef CPPA_DOCUMENTATION
// backward compatiblity, handle_response was part of local_actor
// until version 0.6
sync_handle_helper handle_response(const message_future& mf);
local_actor(bool is_scheduled = false);
virtual bool initialized() = 0;
......
......@@ -39,6 +39,8 @@
#include "cppa/message_id.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/util/callable_trait.hpp"
namespace cppa {
/**
......@@ -77,9 +79,9 @@ class message_future {
* @brief Sets @p mexpr as event-handler for the response message.
*/
template<typename... Cases, typename... Args>
continue_helper then(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check();
self->become_waiting_for(match_expr_convert(arg0, args...), m_mid);
continue_helper then(const match_expr<Cases...>& a0, const Args&... as) {
check_consistency();
self->become_waiting_for(match_expr_convert(a0, as...), m_mid);
return {m_mid};
}
......@@ -88,7 +90,7 @@ class message_future {
*/
template<typename... Cases, typename... Args>
void await(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check();
check_consistency();
self->dequeue_response(match_expr_convert(arg0, args...), m_mid);
}
......@@ -97,11 +99,14 @@ class message_future {
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template<typename F>
typename std::enable_if<util::is_callable<F>::value,continue_helper>::type
then(F fun) {
consistency_check();
self->become_waiting_for(bhvr_from_fun(fun), m_mid);
template<typename... Fs>
typename std::enable_if<
util::all_callable<Fs...>::value,
continue_helper
>::type
then(Fs... fs) {
check_consistency();
self->become_waiting_for(fs2bhvr(std::move(fs)...), m_mid);
return {m_mid};
}
......@@ -110,10 +115,11 @@ class message_future {
* calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template<typename F>
typename std::enable_if<util::is_callable<F>::value>::type await(F fun) {
consistency_check();
self->dequeue_response(bhvr_from_fun(fun), m_mid);
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
check_consistency();
self->dequeue_response(fs2bhvr(std::move(fs)...), m_mid);
}
/**
......@@ -134,28 +140,21 @@ class message_future {
message_id_t m_mid;
template<typename F>
behavior bhvr_from_fun(F fun) {
auto handle_sync_failure = []() -> bool {
self->handle_sync_failure();
return false; // do not treat this as a match to cause a
// continuation to be invoked only in case
// `fun` was invoked
};
template<typename... Fs>
behavior fs2bhvr(Fs... fs) {
auto handle_sync_timeout = []() -> bool {
self->handle_sync_timeout();
return false;
};
return {
on<atom("EXITED"), std::uint32_t>() >> handle_sync_failure,
on<atom("EXITED"),std::uint32_t>() >> skip_message,
on(atom("VOID")) >> skip_message,
on(atom("TIMEOUT")) >> handle_sync_timeout,
on(atom("VOID")) >> handle_sync_failure,
on(any_vals, arg_match) >> fun,
others() >> handle_sync_failure
(on(any_vals, arg_match) >> fs)...
};
}
void consistency_check() {
void check_consistency() {
if (!m_mid.valid() || !m_mid.is_response()) {
throw std::logic_error("handle does not point to a response");
}
......@@ -166,6 +165,62 @@ class message_future {
};
class sync_handle_helper {
public:
inline sync_handle_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Args>
inline message_future::continue_helper operator()(Args&&... args) {
return m_mf.then(std::forward<Args>(args)...);
}
private:
message_future m_mf;
};
class sync_receive_helper {
public:
inline sync_receive_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Args>
inline void operator()(Args&&... args) {
m_mf.await(std::forward<Args>(args)...);
}
private:
message_future m_mf;
};
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_handle_helper handle_response(const message_future& f) {
return {f};
}
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_receive_helper receive_response(const message_future& f) {
return {f};
}
} // namespace cppa
#endif // MESSAGE_FUTURE_HPP
......@@ -32,23 +32,22 @@
#define CPPA_MESSAGE_ID_HPP
#include "cppa/config.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
struct invalid_message_id_flag { constexpr invalid_message_id_flag() { } };
struct invalid_message_id { constexpr invalid_message_id() { } };
/**
* @brief
* @note Asynchronous messages always have an invalid message id.
*/
class message_id_t {
class message_id_t : util::comparable<message_id_t> {
static constexpr std::uint64_t response_flag_mask = 0x8000000000000000;
static constexpr std::uint64_t answered_flag_mask = 0x4000000000000000;
static constexpr std::uint64_t request_id_mask = 0x3FFFFFFFFFFFFFFF;
friend bool operator==(const message_id_t& lhs, const message_id_t& rhs);
public:
constexpr message_id_t() : m_value(0) { }
......@@ -101,9 +100,14 @@ class message_id_t {
return result;
}
static constexpr invalid_message_id_flag invalid = invalid_message_id_flag();
static constexpr invalid_message_id invalid = invalid_message_id{};
constexpr message_id_t(invalid_message_id) : m_value(0) { }
inline message_id_t(invalid_message_id_flag) : m_value(0) { }
long compare(const message_id_t& other) const {
return (m_value == other.m_value)
? 0 : ((m_value < other.m_value) ? -1 : 1);
}
private:
......@@ -113,14 +117,6 @@ class message_id_t {
};
inline bool operator==(const message_id_t& lhs, const message_id_t& rhs) {
return lhs.m_value == rhs.m_value;
}
inline bool operator!=(const message_id_t& lhs, const message_id_t& rhs) {
return !(lhs == rhs);
}
} // namespace cppa
#endif // CPPA_MESSAGE_ID_HPP
......@@ -227,24 +227,29 @@ namespace cppa {
/**
* @brief A wildcard that matches the argument types
* of a given callback. Must be the last argument to {@link on()}.
* of a given callback. Must be the last argument to {@link on()}.
* @see {@link math_actor_example.cpp Math Actor Example}
*/
constexpr ___ arg_match;
constexpr __unspecified__ arg_match;
/**
* @brief Left-hand side of a partial function expression.
*
* Equal to <tt>on(arg_match)</tt>.
*/
constexpr ___ on_arg_match;
constexpr __unspecified__ on_arg_match;
/**
* @brief Right-hand side expression to *not* match a particular pattern.
*/
constexpr __unspecified__ skip_message;
/**
* @brief A wildcard that matches any value of type @p T.
* @see {@link math_actor_example.cpp Math Actor Example}
*/
template<typename T>
___ val();
__unspecified__ val();
/**
* @brief A wildcard that matches any number of any values.
......@@ -259,7 +264,7 @@ constexpr anything any_vals;
* {@link cppa::any_vals any_vals} and {@link cppa::arg_match arg_match}.
*/
template<typename Arg0, typename... Args>
___ on(const Arg0& arg0, const Args&... args);
__unspecified__ on(const Arg0& arg0, const Args&... args);
/**
* @brief Left-hand side of a partial function expression that matches types.
......@@ -268,7 +273,7 @@ ___ on(const Arg0& arg0, const Args&... args);
* can be used as wildcard to match any number of elements of any types.
*/
template<typename... Ts>
___ on();
__unspecified__ on();
/**
* @brief Left-hand side of a partial function expression that matches types.
......@@ -278,10 +283,12 @@ ___ on();
* can be used as wildcard to match any number of elements of any types.
*/
template<atom_value... Atoms, typename... Ts>
___ on();
__unspecified__ on();
#else
bool skip_message();
template<typename T>
constexpr typename detail::boxed<T>::type val() {
return typename detail::boxed<T>::type();
......
......@@ -36,6 +36,8 @@
#include "cppa/util/rm_ref.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/util/conjunction.hpp"
namespace cppa { namespace util {
template<typename Signature>
......@@ -123,6 +125,11 @@ struct is_callable {
};
template<typename... Ts>
struct all_callable {
static constexpr bool value = conjunction<is_callable<Ts>...>::value;
};
template<bool IsCallable, typename C>
struct get_result_type_impl {
typedef typename get_callable_trait<C>::type trait_type;
......
......@@ -38,13 +38,18 @@ namespace cppa { namespace util {
template<typename... BooleanConstants>
struct conjunction;
template<typename T0, typename... Ts>
struct conjunction<T0, Ts...>
: std::integral_constant<bool, T0::value && conjunction<Ts...>::value> {
template<>
struct conjunction<> : std::false_type { };
template<typename T0>
struct conjunction<T0> {
static constexpr bool value = T0::value;
};
template<>
struct conjunction<> : std::true_type { };
template<typename T0, typename T1, typename... Ts>
struct conjunction<T0,T1,Ts...> {
static constexpr bool value = T0::value && conjunction<T1,Ts...>::value;
};
} } // namespace cppa::util
......
No preview for this file type
......@@ -62,7 +62,7 @@ bool abstract_scheduled_actor::chained_enqueue(actor* sender, any_tuple msg) {
}
bool abstract_scheduled_actor::chained_sync_enqueue(actor* sender, message_id_t id, any_tuple msg) {
bool failed;
bool failed = false;
bool result = enqueue_node(super::fetch_node(sender, std::move(msg), id), pending, &failed);
if (failed) {
sync_request_bouncer f{this, exit_reason()};
......@@ -81,7 +81,7 @@ void abstract_scheduled_actor::enqueue(actor* sender, any_tuple msg) {
}
void abstract_scheduled_actor::sync_enqueue(actor* sender, message_id_t id, any_tuple msg) {
bool failed;
bool failed = false;
enqueue_node(super::fetch_node(sender, std::move(msg), id), ready, &failed);
if (failed) {
sync_request_bouncer f{this, exit_reason()};
......
......@@ -29,30 +29,30 @@
#include <iterator>
#include "cppa/local_actor.hpp"
#include "cppa/detail/behavior_stack.hpp"
using namespace std;
namespace cppa { namespace detail {
class behavior_stack_help_iterator
: public std::iterator<std::output_iterator_tag, void, void, void, void> {
struct behavior_stack_mover : iterator<output_iterator_tag,void,void,void,void>{
public:
typedef behavior_stack::element_type element_type;
behavior_stack_mover(behavior_stack* st) : m_stack(st) { }
explicit behavior_stack_help_iterator(behavior_stack& st) : m_stack(&st) { }
behavior_stack_help_iterator& operator=(element_type&& rval) {
m_stack->m_erased_elements.emplace_back(std::move(rval.first));
behavior_stack_mover& operator=(behavior_stack::element_type&& rval) {
m_stack->m_erased_elements.emplace_back(move(rval.first));
return *this;
}
behavior_stack_help_iterator& operator*() { return *this; }
behavior_stack_mover& operator*() { return *this; }
behavior_stack_help_iterator& operator++() { return *this; }
behavior_stack_mover& operator++() { return *this; }
behavior_stack_help_iterator operator++(int) { return *this; }
behavior_stack_mover operator++(int) { return *this; }
private:
......@@ -60,10 +60,12 @@ class behavior_stack_help_iterator
};
inline behavior_stack_mover move_iter(behavior_stack* bs) { return {bs}; }
option<behavior&> behavior_stack::sync_handler(message_id_t expected_response) {
if (expected_response.valid()) {
auto e = m_elements.rend();
auto i = std::find_if(m_elements.rbegin(), e, [=](element_type& val) {
auto i = find_if(m_elements.rbegin(), e, [=](element_type& val) {
return val.second == expected_response;
});
if (i != e) return i->first;
......@@ -71,55 +73,19 @@ option<behavior&> behavior_stack::sync_handler(message_id_t expected_response) {
return {};
}
void behavior_stack::pop_async_back() {
if (m_elements.empty()) {
// nothing to do
}
else if (m_elements.back().second.valid() == false) {
m_erased_elements.emplace_back(std::move(m_elements.back().first));
m_elements.pop_back();
}
else {
auto rlast = m_elements.rend();
auto ri = std::find_if(m_elements.rbegin(), rlast,
[](element_type& e) {
return e.second.valid() == false;
});
if (ri != rlast) {
m_erased_elements.emplace_back(std::move(ri->first));
auto i = ri.base();
--i; // adjust 'normal' iterator to point to the correct element
m_elements.erase(i);
}
}
}
void behavior_stack::erase(message_id_t response_id) {
auto last = m_elements.end();
auto i = std::find_if(m_elements.begin(), last, [=](element_type& e) {
return e.second == response_id;
if (m_elements.empty()) { } // nothing to do
else if (!m_elements.back().second.valid()) erase_at(m_elements.end() - 1);
else rerase_if([](const element_type& e) {
return e.second.valid() == false;
});
if (i != last) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
}
void behavior_stack::push_back(behavior&& what, message_id_t response_id) {
m_elements.emplace_back(std::move(what), response_id);
}
void behavior_stack::clear() {
if (m_elements.empty() == false) {
std::move(m_elements.begin(), m_elements.end(),
behavior_stack_help_iterator{*this});
move(m_elements.begin(), m_elements.end(), move_iter(this));
m_elements.clear();
}
}
void behavior_stack::cleanup() {
m_erased_elements.clear();
}
} } // namespace cppa::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 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/exit_reason.hpp"
namespace cppa { namespace exit_reason {
namespace { constexpr const char* s_names_table[] = {
"not_exited",
"normal",
"unhandled_exception",
"unallowed_function_call",
"unhandled_sync_failure",
"unhandled_sync_timeout"
}; }
const char* as_string(std::uint32_t value) {
if (value <= unhandled_sync_timeout) return s_names_table[value];
if (value == remote_link_unreachable) return "remote_link_unreachable";
if (value >= user_defined) return "user_defined";
return "illegal_exit_reason";
}
} } // namespace cppa::exit_reason
......@@ -142,15 +142,6 @@ void local_actor::forward_message(const actor_ptr& new_receiver) {
}
}
sync_recv_helper local_actor::handle_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->become_waiting_for(std::move(bhvr), mf);
}};
}
response_handle local_actor::make_response_handle() {
auto n = m_current_node;
response_handle result(this, n->sender, n->mid.response_id());
......@@ -171,4 +162,8 @@ void local_actor::exec_behavior_stack() {
// default implementation does nothing
}
sync_handle_helper local_actor::handle_response(const message_future& mf) {
return ::cppa::handle_response(mf);
}
} // namespace cppa
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/on.hpp"
namespace cppa {
bool skip_message() { return false; }
} // namespace cppa
......@@ -57,13 +57,4 @@ void receive_loop(partial_function&& rules) {
receive_loop(tmp);
}
sync_recv_helper receive_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->dequeue_response(bhvr, mf);
}};
}
} // namespace cppa
......@@ -33,15 +33,13 @@ const char* cppa_strip_path(const char* fname) {
}
void cppa_unexpected_message(const char* fname, size_t line_num) {
std::cerr << "ERROR in file " << fname << " on line " << line_num
<< ": unexpected message: "
<< to_string(self->last_dequeued())
<< std::endl;
CPPA_FORMAT_STR(std::cerr, fname, line_num,
"ERROR: unexpected message: "
<< to_string(self->last_dequeued()));
}
void cppa_unexpected_timeout(const char* fname, size_t line_num) {
std::cerr << "ERROR in file " << fname << " on line " << line_num
<< ": unexpected timeout" << std::endl;
CPPA_FORMAT_STR(std::cerr, fname, line_num, "ERROR: unexpected timeout");
}
std::vector<std::string> split(const std::string& str, char delim) {
......
......@@ -48,6 +48,10 @@ inline std::string cppa_stream_arg(const cppa::actor_ptr& ptr) {
return cppa::to_string(ptr);
}
inline std::string cppa_stream_arg(const bool& value) {
return value ? "true" : "false";
}
inline void cppa_passed(const char* fname, int line_number) {
CPPA_FORMAT_STR(std::cout, fname, line_number, "passed");
}
......@@ -113,7 +117,7 @@ inline void cppa_check_value(V1 v1,
} \
else CPPA_PRINT("passed")
#define CPPA_CHECK_EQUAL(rhs_loc, lhs_loc) \
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) \
cppa_check_value((lhs_loc), (rhs_loc), __FILE__, __LINE__)
#define CPPA_CHECK_NOT_EQUAL(rhs_loc, lhs_loc) \
......
......@@ -213,8 +213,9 @@ int client_part(const vector<string_pair>& args) {
forward_to(fwd);
}
);
// shutdown handshake
send(server, atom("farewell"));
receive(on(atom("cu")) >> [] { });
shutdown();
return CPPA_TEST_RESULT();
}
......@@ -329,11 +330,7 @@ int main(int argc, char** argv) {
CPPA_PRINT("test group communication via network (inverted setup)");
spawn5_server(remote_client, true);
self->on_sync_failure([&] {
CPPA_ERROR("unexpected message: "
<< to_string(self->last_dequeued())
<< endl);
});
self->on_sync_failure(CPPA_UNEXPECTED_MSG_CB());
// test forward_to "over network and back"
CPPA_PRINT("test forwarding over network 'and back'");
......@@ -348,9 +345,10 @@ int main(int argc, char** argv) {
);
CPPA_PRINT("wait for a last goodbye");
receive (
on(atom("farewell")) >> [] { }
);
receive(on(atom("farewell")) >> [&] {
send(remote_client, atom("cu"));
CPPA_CHECKPOINT();
});
// wait until separate process (in sep. thread) finished execution
if (run_remote_actor) child.join();
shutdown();
......
......@@ -11,6 +11,14 @@ struct sync_mirror : sb_actor<sync_mirror> {
);
};
// replies to 'f' with 0.0f and to 'i' with 0
struct float_or_int : sb_actor<float_or_int> {
behavior init_state = (
on(atom("f")) >> [] { reply(0.0f); },
on(atom("i")) >> [] { reply(0); }
);
};
struct popular_actor : event_based_actor { // popular actors have a buddy
actor_ptr m_buddy;
popular_actor(const actor_ptr& buddy) : m_buddy(buddy) { }
......@@ -116,16 +124,95 @@ struct D : popular_actor {
}
};
/******************************************************************************\
* test case 3: *
* *
* Client Server Worker *
* | | | *
* | | <---(idle)------ | *
* | ---(request)---> | | *
* | | ---(request)---> | *
* | | |---\ *
* | X | | *
* | |<--/ *
* | <------------(response)------------ | *
* X *
\******************************************************************************/
struct server : event_based_actor {
void init() {
auto die = [=] { quit(exit_reason::user_defined); };
become (
on(atom("idle")) >> [=] {
auto worker = last_sender();
become (
keep_behavior,
on(atom("request")) >> [=] {
forward_to(worker);
unbecome(); // await next idle message
},
on(atom("idle")) >> skip_message,
others() >> die
);
},
on(atom("request")) >> skip_message,
others() >> die
);
}
};
int main() {
CPPA_TEST(test__sync_send);
self->on_sync_failure([] {
CPPA_ERROR("received: " << to_string(self->last_dequeued()));
});
spawn_monitor([&] {
int invocations = 0;
auto foi = spawn_link<float_or_int>();
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()));
});
sync_send(foi, atom("i")).then(
[&](int i) { CPPA_CHECK_EQUAL(i, 0); ++invocations; },
[&](float) { CPPA_UNEXPECTED_MSG(); }
)
.continue_with([&] {
sync_send(foi, atom("f")).then(
[&](int) { CPPA_UNEXPECTED_MSG(); },
[&](float f) { CPPA_CHECK_EQUAL(f, 0); ++invocations; }
);
});
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(invocations, 2);
// provoke invocation of self->handle_sync_failure()
bool sync_failure_called = false;
bool int_handler_called = false;
self->on_sync_failure([&] {
sync_failure_called = true;
});
sync_send(foi, atom("f")).await(
on<int>() >> [&] { int_handler_called = true; }
);
CPPA_CHECK_EQUAL(sync_failure_called, true);
CPPA_CHECK_EQUAL(int_handler_called, false);
self->quit(exit_reason::user_defined);
});
receive (
on(atom("DOWN"), exit_reason::user_defined) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
auto mirror = spawn<sync_mirror>();
bool continuation_called = false;
sync_send(mirror, 42).then(
[](int value) { CPPA_CHECK_EQUAL(value, 42); }
)
.continue_with(
[&] { continuation_called = true; }
);
sync_send(mirror, 42)
.then([](int value) { CPPA_CHECK_EQUAL(value, 42); })
.continue_with([&] { continuation_called = true; });
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(continuation_called, true);
quit_actor(mirror, exit_reason::user_defined);
......@@ -174,6 +261,7 @@ int main() {
// first test: sync error must occur, continuation must not be called
bool timeout_occured = false;
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"));
......@@ -186,6 +274,40 @@ int main() {
quit_actor(c, exit_reason::user_defined);
await_all_others_done();
CPPA_CHECKPOINT();
// test use case 3
spawn_monitor([] { // client
auto s = spawn_link<server>(); // server
auto w = spawn_link([] { // worker
receive_loop(on(atom("request")) >> []{ reply(atom("response")); });
});
// first 'idle', then 'request'
send_as(w, s, atom("idle"));
sync_send(s, atom("request")).await(
on(atom("response")) >> [=] {
CPPA_CHECKPOINT();
CPPA_CHECK_EQUAL(self->last_sender(), w);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
// first 'request', then 'idle'
auto handle = sync_send(s, atom("request"));
send_as(w, s, atom("idle"));
receive_response(handle) (
on(atom("response")) >> [=] {
CPPA_CHECKPOINT();
CPPA_CHECK_EQUAL(self->last_sender(), w);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
send(s, "Ever danced with the devil in the pale moonlight?");
// response: {'EXIT', exit_reason::user_defined}
receive_loop(others() >> CPPA_UNEXPECTED_MSG_CB());
});
receive (
on(atom("DOWN"), exit_reason::user_defined) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_others_done();
CPPA_CHECKPOINT();
shutdown();
......
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