Commit 9c5a7b81 authored by Joseph Noir's avatar Joseph Noir

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

parents 23b305ef 3f781282
......@@ -8,3 +8,6 @@ build-gcc/*
Makefile
bin/*
lib/*
manual/manual.pdf
manual/build-pdf/*
manual/build-html/*
......@@ -3,7 +3,7 @@ project(cppa CXX)
set(LIBCPPA_VERSION_MAJOR 0)
set(LIBCPPA_VERSION_MINOR 8)
set(LIBCPPA_VERSION_PATCH 0)
set(LIBCPPA_VERSION_PATCH 1)
# prohibit in-source builds
if ("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_BINARY_DIR}")
......@@ -24,7 +24,7 @@ if (CMAKE_CXX_FLAGS)
else (CMAKE_CXX_FLAGS)
set(CXXFLAGS_PROVIDED false)
set(CMAKE_CXX_FLAGS "-std=c++11 -Wextra -Wall -pedantic")
set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g -ftemplate-backtrace-limit=0")
set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g")
set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os")
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG")
set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O2 -g")
......
Version 0.8.1
-------------
__2013_10_16__
- GCC 4.7 compatibility
- Fixed handling of partial functions in `match_expr_concat`
- Serialize floating points as IEEE 754
- Removed ftemplate-backtrace-limit option (causes build error in GCC 4.7)
Version 0.8
-----------
......@@ -5,6 +15,15 @@ __2013_10_01__
- Added support for typed actors
- Added `brokers`: actors that encapsulate networking
- timed_* function family now correctly handles priorities
- Fixed monitoring of remote actors
- Added new exit reason `user_shutdown`
- Deprecated `reply` function and recommend returning values instead
- Provide operator-> for optional
- New class `optional_variant` with visitor-based API
- Added manual to Git repository
- Added new libCURL example
- Bugfixes
Version 0.7.2
-------------
......
......@@ -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.8"
PROJECT_NUMBER = "Version 0.8.1"
# The OUTPUT_DIRECTORY tag is used to specify the (relative or absolute)
# base path where the generated documentation will be put.
......
......@@ -36,6 +36,7 @@ cppa/detail/fwd.hpp
cppa/detail/get_behavior.hpp
cppa/detail/group_manager.hpp
cppa/detail/handle.hpp
cppa/detail/ieee_754.hpp
cppa/detail/implicit_conversions.hpp
cppa/detail/matches.hpp
cppa/detail/memory.hpp
......@@ -94,6 +95,7 @@ cppa/io/default_message_queue.hpp
cppa/io/default_peer.hpp
cppa/io/default_peer_acceptor.hpp
cppa/io/default_protocol.hpp
cppa/io/event.hpp
cppa/io/input_stream.hpp
cppa/io/ipv4_acceptor.hpp
cppa/io/ipv4_io_stream.hpp
......@@ -160,6 +162,7 @@ cppa/tuple_cast.hpp
cppa/type_lookup_table.hpp
cppa/typed_actor.hpp
cppa/typed_actor_ptr.hpp
cppa/typed_behavior.hpp
cppa/uniform_type_info.hpp
cppa/unit.hpp
cppa/util/abstract_uniform_type_info.hpp
......@@ -209,8 +212,8 @@ examples/qtsupport/qt_group_chat.cpp
examples/remote_actors/distributed_calculator.cpp
examples/remote_actors/group_chat.cpp
examples/remote_actors/group_server.cpp
examples/remote_actors/protobuf_broker.cpp
examples/remote_actors/protobuf.scala
examples/remote_actors/protobuf_broker.cpp
examples/type_system/announce_1.cpp
examples/type_system/announce_2.cpp
examples/type_system/announce_3.cpp
......@@ -321,5 +324,6 @@ unit_testing/test_serialization.cpp
unit_testing/test_spawn.cpp
unit_testing/test_sync_send.cpp
unit_testing/test_tuple.cpp
unit_testing/test_typed_spawn.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
......@@ -54,6 +54,8 @@ class behavior {
public:
typedef std::function<optional<any_tuple> (any_tuple&)> continuation_fun;
/** @cond PRIVATE */
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
......@@ -108,10 +110,7 @@ class behavior {
* whenever this behavior was successfully applied to
* a message.
*/
behavior add_continuation(const partial_function& fun);
//template<typename F>
//inline behavior add_continuation(F fun);
behavior add_continuation(continuation_fun fun);
private:
......
......@@ -31,6 +31,7 @@
#ifndef BEHAVIOR_IMPL_HPP
#define BEHAVIOR_IMPL_HPP
#include "cppa/atom.hpp"
#include "cppa/optional.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/ref_counted.hpp"
......@@ -43,19 +44,30 @@
namespace cppa {
class partial_function;
typedef optional<any_tuple> bhvr_invoke_result;
} // namespace cppa
namespace cppa { namespace detail {
template<class T> struct is_message_id_wrapper {
template<class U> static char (&test(typename U::message_id_wrapper_tag))[1];
template<class U> static char (&test(...))[2];
static constexpr bool value = sizeof(test<T>(0)) == 1;
};
struct optional_any_tuple_visitor {
inline bhvr_invoke_result operator()() const { return any_tuple{}; }
inline bhvr_invoke_result operator()(const none_t&) const { return none; }
template<typename T>
inline bhvr_invoke_result operator()(T& value) const {
inline bhvr_invoke_result operator()(T& value, typename std::enable_if<not is_message_id_wrapper<T>::value>::type* = 0) const {
return make_any_tuple(std::move(value));
}
template<typename T>
inline bhvr_invoke_result operator()(T& value, typename std::enable_if<is_message_id_wrapper<T>::value>::type* = 0) const {
return make_any_tuple(atom("MESSAGE_ID"), value.get_message_id().integer_value());
}
template<typename... Ts>
inline bhvr_invoke_result operator()(cow_tuple<Ts...>& value) const {
return any_tuple{std::move(value)};
......@@ -168,7 +180,7 @@ class default_behavior_impl : public behavior_impl {
}
typename behavior_impl::pointer copy(const generic_timeout_definition& tdef) const {
return new default_behavior_impl<MatchExpr, std::function<void()> >(m_expr, tdef);
return new default_behavior_impl<MatchExpr, std::function<void()>>(m_expr, tdef);
}
void handle_timeout() { m_fun(); }
......@@ -213,6 +225,11 @@ default_behavior_impl<dummy_match_expr, F>* new_default_behavior(util::duration
typedef intrusive_ptr<behavior_impl> behavior_impl_ptr;
// implemented in partial_function.cpp
behavior_impl_ptr combine(behavior_impl_ptr, const partial_function&);
behavior_impl_ptr combine(const partial_function&, behavior_impl_ptr);
behavior_impl_ptr extract(const partial_function&);
} } // namespace cppa::detail
#endif // BEHAVIOR_IMPL_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/>. *
\******************************************************************************/
/******************************************************************************\
* Based on http://beej.us/guide/bgnet/examples/pack2.c *
\******************************************************************************/
#ifndef IEEE_754_HPP
#define IEEE_754_HPP
namespace cppa { namespace detail {
template<typename T>
struct ieee_754_trait;
template<>
struct ieee_754_trait<float> {
static constexpr std::uint32_t bits = 32;
static constexpr std::uint32_t expbits = 8;
static constexpr float zero = 0.0f;
using packed_type = std::uint32_t;
using signed_packed_type = std::int32_t;
using float_type = float;
};
template<>
struct ieee_754_trait<std::uint32_t> : ieee_754_trait<float> { };
template<>
struct ieee_754_trait<double> {
static constexpr std::uint64_t bits = 64;
static constexpr std::uint64_t expbits = 11;
static constexpr double zero = 0.0;
using packed_type = std::uint64_t;
using signed_packed_type = std::int64_t;
using float_type = double;
};
template<>
struct ieee_754_trait<std::uint64_t> : ieee_754_trait<double> { };
template<typename T>
typename ieee_754_trait<T>::packed_type pack754(T f) {
typedef ieee_754_trait<T> trait;
typedef typename trait::packed_type result_type;
// filter special type
if (f == trait::zero) return 0;
auto significandbits = trait::bits - trait::expbits - 1; // -1 for sign bit
// check sign and begin normalization
result_type sign;
T fnorm;
if (f < 0) {
sign = 1;
fnorm = -f;
}
else {
sign = 0;
fnorm = f;
}
// get the normalized form of f and track the exponent
typename ieee_754_trait<T>::packed_type shift = 0;
while (fnorm >= static_cast<T>(2)) {
fnorm /= static_cast<T>(2);
++shift;
}
while (fnorm < static_cast<T>(1)) {
fnorm *= static_cast<T>(2);
--shift;
}
fnorm = fnorm - static_cast<T>(1);
// calculate the binary form (non-float) of the significand data
result_type significand = fnorm * ((static_cast<result_type>(1) << significandbits) + 0.5f);
// get the biased exponent
auto exp = shift + ((1 << (trait::expbits - 1)) - 1); // shift + bias
// return the final answer
return (sign << (trait::bits - 1)) | (exp << (trait::bits - trait::expbits - 1)) | significand;
}
template<typename T>
typename ieee_754_trait<T>::float_type unpack754(T i) {
typedef ieee_754_trait<T> trait;
typedef typename trait::float_type result_type;
if (i == 0) return trait::zero;
auto significandbits = trait::bits - trait::expbits - 1; // -1 for sign bit
// pull the significand
result_type result = (i & ((static_cast<T>(1) << significandbits) - 1)); // mask
result /= (static_cast<T>(1) << significandbits); // convert back to float
result += static_cast<result_type>(1); // add the one back on
// deal with the exponent
auto bias = (1 << (trait::expbits - 1)) - 1;
typename trait::signed_packed_type shift = ((i >> significandbits) & ((static_cast<typename trait::signed_packed_type>(1) << trait::expbits) - 1)) - bias;
while (shift > 0) {
result *= static_cast<result_type>(2);
--shift;
}
while (shift < 0) {
result /= static_cast<result_type>(2);
++shift;
}
// sign it
result *= (i >> (trait::bits - 1)) & 1 ? -1 : 1;
return result;
}
} } // namespace cppa::detail
#endif // IEEE_754_HPP
......@@ -141,7 +141,7 @@ struct matcher<wildcard_position::in_between, Tuple, T...> {
static constexpr size_t wc_pos = static_cast<size_t>(signed_wc_pos);
static_assert( signed_wc_pos > 0
&& signed_wc_pos < (sizeof...(T) - 1),
&& wc_pos < (sizeof...(T) - 1),
"illegal wildcard position");
static inline bool tmatch(const Tuple& tup) {
......
......@@ -36,6 +36,7 @@
#include <iostream>
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/logging.hpp"
#include "cppa/behavior.hpp"
#include "cppa/to_string.hpp"
......@@ -45,6 +46,9 @@
#include "cppa/partial_function.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/matches.hpp"
#include "cppa/util/scope_guard.hpp"
namespace cppa { namespace detail {
......@@ -344,6 +348,90 @@ class receive_policy {
public:
template<class Client>
inline response_handle fetch_response_handle(Client* cl, int) {
return cl->make_response_handle();
}
template<class Client>
inline response_handle fetch_response_handle(Client*, response_handle& hdl) {
return std::move(hdl);
}
// - extracts response message from handler
// - returns true if fun was successfully invoked
template<class Client, class Fun, class MaybeResponseHandle = int>
optional<any_tuple> invoke_fun(Client* client,
any_tuple& msg,
message_id& mid,
Fun& fun,
MaybeResponseHandle hdl = MaybeResponseHandle{}) {
auto res = fun(msg); // might change mid
if (res) {
//message_header hdr{client, sender, mid.is_request() ? mid.response_id()
// : message_id{}};
if (res->empty()) {
// make sure synchronous requests
// always receive a response
if (mid.is_request() && !mid.is_answered()) {
CPPA_LOGMF(CPPA_WARNING, client,
"actor with ID " << client->id()
<< " did not reply to a "
"synchronous request message");
auto fhdl = fetch_response_handle(client, hdl);
if (fhdl.valid()) fhdl.apply(atom("VOID"));
}
} else {
if ( matches<atom_value, std::uint64_t>(*res)
&& res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) {
auto id = res->template get_as<std::uint64_t>(1);
auto msg_id = message_id::from_integer_value(id);
auto ref_opt = client->bhvr_stack().sync_handler(msg_id);
// calls client->response_handle() if hdl is a dummy
// argument, forwards hdl otherwise to reply to the
// original request message
auto fhdl = fetch_response_handle(client, hdl);
if (ref_opt) {
auto& ref = *ref_opt;
// copy original behavior
behavior cpy = ref;
ref = cpy.add_continuation(
[=](any_tuple& intermediate) -> optional<any_tuple> {
if (!intermediate.empty()) {
// do no use lamba expresion type to
// avoid recursive template intantiaton
behavior::continuation_fun f2 = [=](any_tuple& m) -> optional<any_tuple> {
return std::move(m);
};
auto mutable_mid = mid;
// recursively call invoke_fun on the
// result to correctly handle stuff like
// sync_send(...).then(...).then(...)...
return invoke_fun(client,
intermediate,
mutable_mid,
f2,
fhdl);
}
return none;
}
);
}
// reset res to prevent "outer" invoke_fun
// from handling the result again
res->reset();
} else {
// respond by using the result of 'fun'
auto fhdl = fetch_response_handle(client, hdl);
if (fhdl.valid()) fhdl.apply(std::move(*res));
}
}
return res;
}
// fun did not return a value => no match
return none;
}
// workflow 'template'
template<class Client, class Fun, class Policy>
......@@ -392,8 +480,15 @@ class receive_policy {
case sync_response: {
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_LOGMF(CPPA_WARNING, client, "sync failure occured");
auto res = invoke_fun(client,
node->msg,
node->mid,
fun);
if (!res && handle_sync_failure_on_mismatch) {
CPPA_LOGMF(CPPA_WARNING,
client,
"sync failure occured in actor with ID "
<< client->id());
client->handle_sync_failure();
}
client->mark_arrived(awaited_response);
......@@ -406,25 +501,11 @@ class receive_policy {
case ordinary_message: {
if (!awaited_response.valid()) {
auto previous_node = hm_begin(client, node, policy);
auto res = fun(node->msg);
auto res = invoke_fun(client,
node->msg,
node->mid,
fun);
if (res) {
auto mid = node->mid;
auto sender = node->sender;
message_header hdr{client, sender, mid.response_id()};
if (res->empty()) {
// make sure synchronous requests
// always receive a response
if (mid.valid() && !mid.is_answered() && sender) {
CPPA_LOGMF(CPPA_WARNING, client,
"actor did not reply to a "
"synchronous request message");
sender->enqueue(std::move(hdr),
make_any_tuple(atom("VOID")));
}
} else {
// respond by using the result of 'fun'
sender->enqueue(std::move(hdr), std::move(*res));
}
hm_cleanup(client, previous_node, policy);
return hm_msg_handled;
}
......
......@@ -108,7 +108,7 @@ inline void assert_types() {
template<typename T>
struct lifted_result_type {
typedef util::type_list<T> type;
typedef util::type_list<typename detail::implicit_conversions<T>::type> type;
};
template<typename... Ts>
......
......@@ -53,7 +53,8 @@ static constexpr std::uint32_t unhandled_exception = 0x00002;
/**
* @brief Indicates that an event-based actor
* tried to use receive().
* tried to use {@link receive()} or a strongly typed actor tried
* to call {@link become()}.
*/
static constexpr std::uint32_t unallowed_function_call = 0x00003;
......@@ -68,6 +69,12 @@ static constexpr std::uint32_t unhandled_sync_failure = 0x00004;
*/
static constexpr std::uint32_t unhandled_sync_timeout = 0x00005;
/**
* @brief Indicates that the actor was forced to shutdown by
* a user-generated event.
*/
static constexpr std::uint32_t user_shutdown = 0x00010;
/**
* @brief Indicates that an actor finishied execution
* because a connection to a remote link was
......
......@@ -35,6 +35,8 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/io/event.hpp"
namespace cppa { namespace io {
/**
......@@ -102,9 +104,11 @@ class continuable {
/**
* @brief Called from middleman before it removes this object
* due to an IO failure.
* due to an IO failure, can be called twice
* (for read and for write error).
* @param bitmask Either @p read or @p write.
*/
virtual void io_failed() = 0;
virtual void io_failed(event_bitmask bitmask) = 0;
protected:
......@@ -122,7 +126,6 @@ class continuable {
* inline and template member function implementations *
******************************************************************************/
inline native_socket_type continuable::read_handle() const {
return m_rd;
}
......
......@@ -55,7 +55,7 @@ class default_message_queue : public ref_counted {
inline value_type pop() {
value_type result(std::move(m_impl.front()));
m_impl.erase(m_impl.begin());
return std::move(result);
return result;
}
private:
......
......@@ -71,7 +71,7 @@ class default_peer : public extend<continuable>::with<buffered_writing> {
void dispose() override;
void io_failed() override;
void io_failed(event_bitmask mask) override;
void enqueue(const message_header& hdr, const any_tuple& msg);
......
......@@ -56,7 +56,7 @@ class default_peer_acceptor : public continuable {
void dispose() override;
void io_failed() override;
void io_failed(event_bitmask) override;
private:
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 EVENT_HPP
#define EVENT_HPP
namespace cppa { namespace io {
typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace <anonymous>::event
template<unsigned InputEvent, unsigned OutputEvent, unsigned ErrorEvent>
inline event_bitmask from_int_bitmask(unsigned mask) {
event_bitmask result = event::none;
// read/write as long as possible
if (mask & InputEvent) result = event::read;
if (mask & OutputEvent) result |= event::write;
if (result == event::none && mask & ErrorEvent) result = event::error;
return result;
}
} } // namespace cppa::io
#endif // EVENT_HPP
......@@ -37,32 +37,11 @@
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/io/event.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace io {
typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace event
template<unsigned InputEvent, unsigned OutputEvent, unsigned ErrorEvent>
inline event_bitmask from_int_bitmask(unsigned mask) {
event_bitmask result = event::none;
// read/write as long as possible
if (mask & InputEvent) result = event::read;
if (mask & OutputEvent) result |= event::write;
if (result == event::none && mask & ErrorEvent) result = event::error;
return result;
}
struct fd_meta_info {
native_socket_type fd;
continuable* ptr;
......
......@@ -314,7 +314,7 @@ class local_actor : public extend<actor>::with<memory_cached> {
void reply_message(any_tuple&& what);
void forward_message(const actor_ptr& new_receiver);
void forward_message(const actor_ptr& new_receiver, message_priority prio);
inline const actor_ptr& chained_actor();
......
......@@ -115,7 +115,7 @@ inline actor_ptr fwd_aptr(const self_type& s) {
}
inline actor_ptr fwd_aptr(actor_ptr ptr) {
return std::move(ptr);
return ptr;
}
struct oss_wr {
......
......@@ -308,7 +308,7 @@ namespace cppa {
* @returns A helper object providing <tt>operator(...)</tt>.
*/
inline detail::match_helper match(any_tuple what) {
return std::move(what);
return what;
}
/**
......
......@@ -501,7 +501,7 @@ Result unroll_expr(PPFPs& fs,
/* recursively evaluate sub expressions */ {
Result res = unroll_expr<Result>(fs, bitmask, long_constant<N-1>{},
type_token, is_dynamic, ptr, tup);
if (res) return std::move(res);
if (res) return res;
}
if ((bitmask & (0x01 << N)) == 0) return none;
auto& f = get<N>(fs);
......@@ -520,8 +520,8 @@ Result unroll_expr(PPFPs& fs,
}
// PPFP = projection_partial_function_pair
template<class PPFPs, class Tuple>
inline bool can_unroll_expr(PPFPs&, minus1l, const std::type_info&, const Tuple&) {
template<class PPFPs, class T>
inline bool can_unroll_expr(PPFPs&, minus1l, const std::type_info&, const T&) {
return false;
}
......@@ -597,7 +597,7 @@ inline any_tuple& detach_if_needed(any_tuple& tup, std::true_type) {
inline any_tuple detach_if_needed(const any_tuple& tup, std::true_type) {
any_tuple cpy{tup};
cpy.force_detach();
return std::move(cpy);
return cpy;
}
inline const any_tuple& detach_if_needed(const any_tuple& tup, std::false_type) {
......@@ -906,21 +906,40 @@ match_expr_collect(const T& arg, const Ts&... args) {
namespace detail {
typedef std::true_type with_timeout;
typedef std::false_type without_timeout;
// with timeout
//typedef std::true_type with_timeout;
//typedef std::false_type without_timeout;
// end of recursion
template<class Data, class Token>
behavior_impl_ptr concat_rec(const Data& data, Token) {
typedef typename match_expr_from_type_list<Token>::type combined_type;
auto lvoid = [] { };
typedef default_behavior_impl<combined_type, decltype(lvoid)> impl_type;
return new impl_type(data, util::duration{}, lvoid);
}
// end of recursion with nothing but a partial function
inline behavior_impl_ptr concat_rec(const tdata<>&,
util::empty_type_list,
const partial_function& pfun) {
return extract(pfun);
}
// end of recursion with timeout
template<class Data, class Token, typename F>
behavior_impl* concat_rec(const Data& data, Token, const timeout_definition<F>& arg) {
behavior_impl_ptr concat_rec(const Data& data,
Token,
const timeout_definition<F>& arg) {
typedef typename match_expr_from_type_list<Token>::type combined_type;
return new default_behavior_impl<combined_type, F>{data, arg};
}
// recursive concatenation function
template<class Data, class Token, typename T, typename... Ts>
behavior_impl* concat_rec(const Data& data, Token, const T& arg, const Ts&... args) {
behavior_impl_ptr concat_rec(const Data& data,
Token,
const T& arg,
const Ts&... args) {
typedef typename util::tl_concat<
Token,
typename T::cases_list
......@@ -938,57 +957,41 @@ behavior_impl* concat_rec(const Data& data, Token, const T& arg, const Ts&... ar
return concat_rec(next_data, next_token, args...);
}
template<typename F>
behavior_impl* concat_expr(with_timeout, const timeout_definition<F>& arg) {
typedef default_behavior_impl<dummy_match_expr, F> impl_type;
return new impl_type(dummy_match_expr{}, arg);
// handle partial functions at end of recursion
template<class Data, class Token>
behavior_impl_ptr concat_rec(const Data& data,
Token token,
const partial_function& pfun) {
return combine(concat_rec(data, token), pfun);
}
template<typename T, typename... Ts>
behavior_impl* concat_expr(with_timeout, const T& arg, const Ts&... args) {
typename tdata_from_type_list<
typename util::tl_map<
typename T::cases_list,
gref_wrapped
>::type
>::type
wrapper;
detail::rebind_tdata(wrapper, arg.cases());
return concat_rec(wrapper, typename T::cases_list{}, args...);
// handle partial functions in between
template<class Data, class Token, typename T, typename... Ts>
behavior_impl_ptr concat_rec(const Data& data,
Token token,
const partial_function& pfun,
const T& arg,
const Ts&... args) {
auto lhs = concat_rec(data, token);
detail::tdata<> dummy;
auto rhs = concat_rec(dummy, util::empty_type_list{}, arg, args...);
return combine(lhs, pfun)->or_else(rhs);
}
// without timeout
// handle partial functions at recursion start
template<typename T, typename... Ts>
behavior_impl* concat_expr(without_timeout, const T& arg, const Ts&... args) {
typename tdata_from_type_list<
typename util::tl_map<
typename util::tl_concat<
typename T::cases_list,
typename Ts::cases_list...
>::type,
gref_wrapped
>::type
>::type
all_cases;
typedef typename match_expr_from_type_list<
typename util::tl_concat<
typename T::cases_list,
typename Ts::cases_list...
>::type
>::type
combined_type;
auto lvoid = [] { };
typedef default_behavior_impl<combined_type, decltype(lvoid)> impl_type;
rebind_tdata(all_cases, arg.cases(), args.cases()...);
return new impl_type(all_cases, util::duration{}, lvoid);
behavior_impl_ptr concat_rec(const tdata<>& data,
util::empty_type_list token,
const partial_function& pfun,
const T& arg,
const Ts&... args) {
return combine(pfun, concat_rec(data, token, arg, args...));
}
template<typename T, typename... Ts>
behavior_impl_ptr match_expr_concat(const T& arg, const Ts&... args) {
std::integral_constant<bool, util::disjunction<T::may_have_timeout, Ts::may_have_timeout...>::value> token;
// use static call dispatch to select correct function
return concat_expr(token, arg, args...);
detail::tdata<> dummy;
return concat_rec(dummy, util::empty_type_list{}, arg, args...);
}
} // namespace detail
......
......@@ -47,6 +47,8 @@
namespace cppa {
namespace detail { struct optional_any_tuple_visitor; }
/**
* @brief Provides the @p continue_with member function as used in
* <tt>sync_send(...).then(...).continue_with(...)</tt>.
......@@ -55,18 +57,31 @@ class continue_helper {
public:
typedef int message_id_wrapper_tag;
inline continue_helper(message_id mid) : m_mid(mid) { }
template<typename F>
void continue_with(F fun) {
continue_helper& continue_with(F fun) {
return continue_with(behavior::continuation_fun{partial_function{
on(any_vals, arg_match) >> fun
}});
}
inline continue_helper& continue_with(behavior::continuation_fun fun) {
auto ref_opt = self->bhvr_stack().sync_handler(m_mid);
if (ref_opt) {
auto& ref = *ref_opt;
// copy original behavior
behavior cpy = ref;
ref = cpy.add_continuation(on(any_vals, arg_match) >> fun);
ref = cpy.add_continuation(std::move(fun));
}
else CPPA_LOG_ERROR(".continue_with: failed to add continuation");
return *this;
}
inline message_id get_message_id() const {
return m_mid;
}
private:
......@@ -175,14 +190,52 @@ class typed_continue_helper {
public:
typedef int message_id_wrapper_tag;
typedef typename detail::lifted_result_type<R>::type result_types;
typed_continue_helper(continue_helper ch) : m_ch(std::move(ch)) { }
template<typename F>
void continue_with(F fun) {
typed_continue_helper<typename util::get_callable_trait<F>::result_type>
continue_with(F fun) {
detail::assert_types<result_types, F>();
m_ch.continue_with(std::move(fun));
return {m_ch};
}
inline message_id get_message_id() const {
return m_ch.get_message_id();
}
private:
continue_helper m_ch;
};
/*
template<>
class typed_continue_helper<void> {
public:
typedef int message_id_wrapper_tag;
typedef util::type_list<void> result_types;
typed_continue_helper(continue_helper ch) : m_ch(std::move(ch)) { }
template<typename F>
void continue_with(F fun) {
using arg_types = typename util::get_callable_trait<F>::arg_types;
static_assert(std::is_same<util::empty_type_list, arg_types>::value,
"functor takes too much arguments");
m_ch.continue_with(std::move(fun));
}
inline message_id get_message_id() const {
return m_ch.get_message_id();
}
private:
......@@ -190,6 +243,7 @@ class typed_continue_helper {
continue_helper m_ch;
};
*/
template<typename OutputList>
class typed_message_future {
......
......@@ -69,6 +69,9 @@ struct optional_variant_copy_helper {
inline void operator()() const {
lhs = unit;
}
inline void operator()(const none_t&) const {
lhs = none;
}
};
template<typename T>
......@@ -448,7 +451,7 @@ make_optional_variant(T value, Ts&&... args) {
template<typename... Ts>
inline optional_variant<Ts...> make_optional_variant(optional_variant<Ts...> value) {
return std::move(value);
return value;
}
} // namespace cppa
......
......@@ -65,7 +65,7 @@ class partial_function {
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
inline auto as_behavior_impl() const -> const impl_ptr&;
inline auto as_behavior_impl() const -> impl_ptr;
partial_function(impl_ptr ptr);
......@@ -142,7 +142,7 @@ match_expr_convert(const T0& arg0, const T1& arg1, const Ts&... args) {
template<typename... Cases>
partial_function operator,(const match_expr<Cases...>& mexpr,
const partial_function& pfun) {
return mexpr.as_behavior_impl()->or_else(pfun.as_behavior_impl);
return mexpr.as_behavior_impl()->or_else(pfun.as_behavior_impl());
}
/******************************************************************************
......@@ -177,7 +177,7 @@ partial_function::or_else(Ts&&... args) const {
return m_impl->or_else(tmp.as_behavior_impl());
}
inline auto partial_function::as_behavior_impl() const -> const impl_ptr& {
inline auto partial_function::as_behavior_impl() const -> impl_ptr {
return m_impl;
}
......
......@@ -108,35 +108,26 @@ class scheduler {
virtual attachable* register_hidden_context();
template<typename Duration, typename... Data>
void delayed_send(const channel_ptr& to,
void delayed_send(message_header hdr,
const Duration& rel_time,
any_tuple data ) {
auto tup = make_any_tuple(atom("SEND"),
util::duration{rel_time},
to,
std::move(hdr),
std::move(data));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
template<typename Duration, typename... Data>
void delayed_reply(const actor_ptr& to,
void delayed_reply(message_header hdr,
const Duration& rel_time,
message_id id,
any_tuple data ) {
CPPA_REQUIRE(!id.valid() || id.is_response());
if (id.valid()) {
auto tup = make_any_tuple(atom("REPLY"),
CPPA_REQUIRE(hdr.id.valid() && hdr.id.is_response());
auto tup = make_any_tuple(atom("SEND"),
util::duration{rel_time},
to,
id,
std::move(hdr),
std::move(data));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
}
else {
this->delayed_send(to, rel_time, std::move(data));
}
delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
/**
......
This diff is collapsed.
......@@ -49,7 +49,7 @@ inline actor_ptr eval_sopts(spawn_options opts, local_actor_ptr ptr) {
<< " of type " << detail::demangle(typeid(*ptr)));
if (has_monitor_flag(opts)) self->monitor(ptr);
if (has_link_flag(opts)) self->link_to(ptr);
return std::move(ptr);
return ptr;
}
/** @endcond */
......@@ -126,35 +126,43 @@ actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns.
*/
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
template<class Impl, spawn_options Options, typename... Ts>
actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
ptr->join(grp);
return eval_sopts(Options, get_scheduler()->exec(Options, ptr));
}
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
typename Impl::typed_pointer_type spawn_typed(Ts&&... args) {
static_assert(util::tl_is_distinct<typename Impl::signatures>::value,
"typed actors are not allowed to define "
"multiple patterns with identical signature");
auto p = make_counted<Impl>(std::forward<Ts>(args)...);
using result_type = typename Impl::typed_pointer_type;
return result_type::cast_from(
eval_sopts(Options, get_scheduler()->exec(Options, std::move(p)))
);
}
template<spawn_options Options, typename... Ts>
typed_actor_ptr<typename detail::deduce_signature<Ts>::type...>
spawn_typed(const match_expr<Ts...>& me) {
static_assert(util::conjunction<
detail::match_expr_has_no_guard<Ts>::value...
>::value,
"typed actors are not allowed to use guard expressions");
typedef util::type_list<
static_assert(util::tl_is_distinct<
util::type_list<
typename detail::deduce_signature<Ts>::arg_types...
>
args;
static_assert(util::tl_is_distinct<args>::value,
"typed actors are not allowed to define multiple patterns "
"with identical signature");
auto ptr = make_counted<typed_actor<match_expr<Ts...>>>(me);
return {eval_sopts(Options, get_scheduler()->exec(Options, std::move(ptr)))};
>::value,
"typed actors are not allowed to define "
"multiple patterns with identical signature");
using impl = detail::default_typed_actor<
typename detail::deduce_signature<Ts>::type...
>;
return spawn_typed<impl, Options>(me);
}
template<typename... Ts>
......
......@@ -32,6 +32,7 @@
#define CPPA_TYPED_ACTOR_HPP
#include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/message_future.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -39,75 +40,60 @@
namespace cppa {
struct typed_actor_result_visitor {
template<typename... Signatures>
class typed_actor_ptr;
typed_actor_result_visitor() : m_hdl(self->make_response_handle()) { }
template<typename... Signatures>
class typed_actor : public event_based_actor {
typed_actor_result_visitor(response_handle hdl) : m_hdl(hdl) { }
public:
inline void operator()(const none_t&) const {
CPPA_LOG_ERROR("a typed actor received a "
"non-matching message: "
<< to_string(self->last_dequeued()));
}
using signatures = util::type_list<Signatures...>;
inline void operator()() const { }
using behavior_type = typed_behavior<Signatures...>;
template<typename T>
inline void operator()(T& value) const {
reply_to(m_hdl, std::move(value));
}
using typed_pointer_type = typed_actor_ptr<Signatures...>;
protected:
virtual behavior_type make_behavior() = 0;
template<typename... Ts>
inline void operator()(cow_tuple<Ts...>& value) const {
reply_tuple_to(m_hdl, std::move(value));
void init() final {
auto bhvr = make_behavior();
m_bhvr_stack.push_back(std::move(bhvr.unbox()));
}
template<typename R>
inline void operator()(typed_continue_helper<R>& ch) const {
auto hdl = m_hdl;
ch.continue_with([=](R value) {
typed_actor_result_visitor tarv{hdl};
auto ov = make_optional_variant(std::move(value));
apply_visitor(tarv, ov);
});
void do_become(behavior&&, bool) final {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
}
private:
};
response_handle m_hdl;
} // namespace cppa
};
namespace cppa { namespace detail {
template<typename MatchExpr>
class typed_actor : public event_based_actor {
template<typename... Signatures>
class default_typed_actor : public typed_actor<Signatures...> {
public:
typed_actor(MatchExpr expr) : m_fun(std::move(expr)) { }
template<typename... Cases>
default_typed_actor(match_expr<Cases...> expr) : m_bhvr(std::move(expr)) { }
protected:
void init() override {
m_bhvr_stack.push_back(partial_function{
on<anything>() >> [=] {
auto result = m_fun(last_dequeued());
apply_visitor(typed_actor_result_visitor{}, result);
}
});
}
void do_become(behavior&&, bool) override {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
typed_behavior<Signatures...> make_behavior() override {
return m_bhvr;
}
private:
MatchExpr m_fun;
typed_behavior<Signatures...> m_bhvr;
};
} // namespace cppa
} } // namespace cppa::detail
#endif // CPPA_TYPED_ACTOR_HPP
......@@ -38,67 +38,9 @@
namespace cppa {
template<typename... Signatures>
class typed_actor_ptr;
// functions that need access to typed_actor_ptr::unbox()
template<spawn_options Options, typename... Ts>
typed_actor_ptr<typename detail::deduce_signature<Ts>::type...>
spawn_typed(const match_expr<Ts...>&);
template<typename... Ts>
void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>&, Ts&&...);
template<typename... Ts, typename... Us>
typed_message_future<
typename detail::deduce_output_type<
util::type_list<Ts...>,
util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Us
>::type
>::type...
>
>::type
>
sync_send(const typed_actor_ptr<Ts...>&, Us&&...);
template<typename... Signatures>
class typed_actor_ptr {
template<typename... OtherSignatures>
friend class typed_actor_ptr;
template<spawn_options Options, typename... Ts>
friend typed_actor_ptr<typename detail::deduce_signature<Ts>::type...>
spawn_typed(const match_expr<Ts...>&);
template<typename... Ts>
friend void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Sigs, typename... Ts>
friend void send(const typed_actor_ptr<Sigs...>&, Ts&&...);
template<typename... Ts, typename... Us>
friend typed_message_future<
typename detail::deduce_output_type<
util::type_list<Ts...>,
util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Us
>::type
>::type...
>
>::type
>
sync_send(const typed_actor_ptr<Ts...>&, Us&&...);
public:
typedef util::type_list<Signatures...> signatures;
......@@ -111,22 +53,39 @@ class typed_actor_ptr {
template<typename... Others>
typed_actor_ptr(typed_actor_ptr<Others...> other) {
static_assert(util::tl_is_strict_subset<
util::type_list<Signatures...>,
util::type_list<Others...>
>::value,
"'this' must be a strict subset of 'other'");
m_ptr = std::move(other.m_ptr);
set(std::move(other));
}
private:
template<typename... Others>
typed_actor_ptr& operator=(typed_actor_ptr<Others...> other) {
set(std::move(other));
return *this;
}
/** @cond PRIVATE */
const actor_ptr& unbox() const { return m_ptr; }
static typed_actor_ptr cast_from(actor_ptr from) {
return {std::move(from)};
}
const actor_ptr& type_erased() const { return m_ptr; }
actor_ptr& type_erased() { return m_ptr; }
/** @endcond */
private:
typed_actor_ptr(actor_ptr ptr) : m_ptr(std::move(ptr)) { }
template<class ListA, class ListB>
inline void check_signatures() {
static_assert(util::tl_is_strict_subset<ListA, ListB>::value,
"'this' must be a strict subset of 'other'");
}
template<typename... Others>
inline void set(typed_actor_ptr<Others...> other) {
check_signatures<signatures, util::type_list<Others...>>();
m_ptr = std::move(other.type_erased());
}
actor_ptr m_ptr;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 TYPED_BEHAVIOR_HPP
#define TYPED_BEHAVIOR_HPP
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/detail/typed_actor_util.hpp"
namespace cppa {
template<typename... Signatures>
class typed_actor;
template<typename... Signatures>
class typed_behavior {
template<typename... OtherSignatures>
friend class typed_actor;
typed_behavior() = delete;
public:
typed_behavior(typed_behavior&&) = default;
typed_behavior(const typed_behavior&) = default;
typed_behavior& operator=(typed_behavior&&) = default;
typed_behavior& operator=(const typed_behavior&) = default;
typedef util::type_list<Signatures...> signatures;
template<typename... Cs>
typed_behavior(match_expr<Cs...> expr) {
set(std::move(expr));
}
template<typename... Cs>
typed_behavior& operator=(match_expr<Cs...> expr) {
set(std::move(expr));
return *this;
}
private:
behavior& unbox() { return m_bhvr; }
template<typename... Cs>
void set(match_expr<Cs...>&& expr) {
m_bhvr = std::move(expr);
using input = util::type_list<
typename detail::deduce_signature<Cs>::type...
>;
// check whether the signature is an exact match
static_assert(util::tl_equal<signatures, input>::value,
"'expr' does not match given signature");
}
behavior m_bhvr;
};
} // namespace cppa
#endif // TYPED_BEHAVIOR_HPP
......@@ -82,6 +82,7 @@ class duration {
"only seconds, milliseconds or microseconds allowed");
}
// convert minutes to seconds
template<class Rep>
constexpr duration(std::chrono::duration<Rep, std::ratio<60, 1> > d)
: unit(time_unit::seconds), count(d.count() * 60) { }
......
......@@ -1052,6 +1052,17 @@ struct tl_is_strict_subset {
>::value;
};
/**
* @brief Tests whether ListA equal to ListB.
*/
template<class ListA, class ListB>
struct tl_equal {
static constexpr bool value =
tl_is_strict_subset<ListA, ListB>::value
&& tl_is_strict_subset<ListB, ListA>::value;
};
/**
* @}
*/
......
......@@ -394,7 +394,7 @@ int main() {
while (!shutdown_flag) { sleep(1); }
aout << color::cyan << "received CTRL+C" << color::reset_endl;
// shutdown actors
send_exit(master, exit_reason::user_defined);
send_exit(master, exit_reason::user_shutdown);
// await actors
act.sa_handler = [](int) { abort(); };
set_sighandler();
......
......@@ -55,7 +55,7 @@ void tester(const actor_ptr& testee) {
// will be invoked if we receive an unexpected response message
self->on_sync_failure([] {
aout << "AUT (actor under test) failed" << endl;
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
});
// first test: 2 + 1 = 3
sync_send(testee, atom("plus"), 2, 1).then(
......
......@@ -46,7 +46,7 @@ inline string trim(std::string s) {
s.erase(s.begin(), find_if(s.begin(), s.end(), not_space));
// trim right
s.erase(find_if(s.rbegin(), s.rend(), not_space).base(), s.end());
return std::move(s);
return s;
}
void client_bhvr(const string& host, uint16_t port, const actor_ptr& server) {
......@@ -112,7 +112,7 @@ void client_repl(const string& host, uint16_t port) {
line = trim(std::move(line)); // ignore leading and trailing whitespaces
if (line == "quit") {
// force client to quit
send_exit(client, exit_reason::user_defined);
send_exit(client, exit_reason::user_shutdown);
return;
}
// the STL way of line.starts_with("connect")
......
......@@ -142,7 +142,7 @@ int main(int argc, char** argv) {
}
);
// force actor to quit
send_exit(client_actor, exit_reason::user_defined);
send_exit(client_actor, exit_reason::user_shutdown);
await_all_others_done();
shutdown();
return 0;
......
......@@ -122,7 +122,7 @@ void protobuf_io(broker* thisptr, connection_handle hdl, const actor_ptr& buddy)
send(buddy, atom("pong"), p.pong().id());
}
else {
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
cerr << "neither Ping nor Pong!" << endl;
}
// receive next length prefix
......@@ -138,7 +138,7 @@ void protobuf_io(broker* thisptr, connection_handle hdl, const actor_ptr& buddy)
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
aout << "someone is trying something nasty" << endl;
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
return;
}
// receive protobuf data
......
No preview for this file type
......@@ -67,6 +67,8 @@ To obtain a string representation of an error code, use \lstinline^cppa::exit_re
\hline
\lstinline^unhandled_sync_timeout^ & 5 & Actor was killed, because no timeout handler was set and a synchronous message timed out \\
\hline
\lstinline^user_shutdown^ & 16 & Actor was killed by a user-generated event \\
\hline
\lstinline^remote_link_unreachable^ & 257 & Indicates that a remote actor became unreachable, e.g., due to connection error \\
\hline
\lstinline^user_defined^ & 65536 & Minimum value for user-defined exit codes \\
......
......@@ -50,7 +50,7 @@ class local_actor;
\hline
\lstinline^any_tuple last_dequeued()^ & Returns the last message that was dequeued from the actor's mailbox\newline\textbf{Note}: Only set during callback invocation \\
\hline
\lstinline^actor_ptr last_sender()^ & Returns the sender of the last dequeued message\newline\textbf{Note$_{1}$}: Only set during callback invocation\newline\textbf{Note$_{2}$}: Used by the function \lstinline^reply^ (see Section \ref{Sec::Send::Reply}) \\
\lstinline^actor_ptr last_sender()^ & Returns the sender of the last dequeued message\newline\textbf{Note$_{1}$}: Only set during callback invocation\newline\textbf{Note$_{2}$}: Used implicitly to send response messages (see Section \ref{Sec::Send::Reply}) \\
\hline
\lstinline^vector<group_ptr> joined_groups()^ & Returns all subscribed groups \\
\hline
......
......@@ -25,13 +25,12 @@ Related functions, e.g., \lstinline^sync_send(...).then(...)^, behave the same,
\begin{itemize}
\item
\lstinline^send(self->last_sender(), ...)^ is \textbf{not} equal to \lstinline^reply(...)^.
The two functions \lstinline^receive_response^ and \lstinline^handle_response^ will only recognize messages send via either \lstinline^reply^ or \lstinline^reply_tuple^.
\lstinline^send(self->last_sender(), ...)^ does \textbf{not} send a response message.
\item
A handle returned by \lstinline^sync_send^ represents \emph{exactly one} response message.
Therefore, it is not possible to receive more than one response message.
Calling \lstinline^reply^ more than once will result in lost messages and calling \lstinline^handle_response^ or \lstinline^receive_response^ more than once on a future will throw an exception.
%Calling \lstinline^reply^ more than once will result in lost messages and calling \lstinline^handle_response^ or \lstinline^receive_response^ more than once on a future will throw an exception.
\item
The future returned by \lstinline^sync_send^ is bound to the calling actor.
......
......@@ -66,13 +66,14 @@ void mirror() {
// wait for messages
become (
// invoke this lambda expression if we receive a string
on_arg_match >> [](const string& what) {
on_arg_match >> [](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper)
aout << what << endl;
// replies "!dlroW olleH"
reply(string(what.rbegin(), what.rend()));
// terminates this actor ('become' otherwise loops forever)
// terminates this actor afterwards;
// 'become' otherwise loops forever
self->quit();
// replies "!dlroW olleH"
return string(what.rbegin(), what.rend());
}
);
}
......
......@@ -18,7 +18,7 @@ Otherwise, the actor accepts all incoming connections (\lstinline^INADDR_ANY^).
publish(self, 4242);
become (
on(atom("ping"), arg_match) >> [](int i) {
reply(atom("pong"), i);
return make_cow_tuple(atom("pong"), i);
}
);
\end{lstlisting}
......@@ -40,7 +40,7 @@ become (
self->quit();
},
on(atom("pong"), arg_match) >> [](int i) {
reply(atom("ping"), i+1);
return make_cow_tuple(atom("ping"), i+1);
}
);
\end{lstlisting}
......@@ -39,55 +39,59 @@ struct printer : sb_actor<printer> {
Note that \lstinline^sb_actor^ uses the Curiously Recurring Template Pattern. Thus, the derived class must be given as template parameter.
This technique allows \lstinline^sb_actor^ to access the \lstinline^init_state^ member of a derived class.
The following example illustrates a more advanced state-based actor that implements a stack with a fixed maximum number of elements.
Note that this example uses non-static member initialization and thus might not compile with some compilers.
\clearpage
\begin{lstlisting}
class fixed_stack : public sb_actor<fixed_stack> {
// grant access to the private init_state member
friend class sb_actor<fixed_stack>;
static constexpr size_t max_size = 10;
size_t max_size = 10;
std::vector<int> data;
vector<int> data;
behavior empty = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
behavior full;
behavior filled;
behavior empty;
behavior& init_state = empty;
public:
fixed_stack(size_t max) : max_size(max) {
full = (
on(atom("push"), arg_match) >> [=](int) { /* discard */ },
on(atom("pop")) >> [=]() -> cow_tuple<atom_value, int> {
auto result = data.back();
data.pop_back();
become(filled);
},
on(atom("pop")) >> [=]() {
reply(atom("failure"));
return {atom("ok"), result};
}
);
behavior filled = (
filled = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
if (data.size() == max_size)
become(full);
if (data.size() == max_size) become(full);
},
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
on(atom("pop")) >> [=]() -> cow_tuple<atom_value, int> {
auto result = data.back();
data.pop_back();
if (data.empty())
become(empty);
if (data.empty()) become(empty);
return {atom("ok"), result};
}
);
behavior full = (
on(atom("push"), arg_match) >> [=](int) { },
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
data.pop_back();
empty = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
become(filled);
},
on(atom("pop")) >> [=] {
return atom("failure");
}
);
behavior& init_state = empty;
}
};
\end{lstlisting}
......
......@@ -35,24 +35,18 @@ Choosing one or the other is merely a matter of personal preferences.
\subsection{Replying to Messages}
\label{Sec::Send::Reply}
The return value of a message handler is used as response message.
During callback invokation, \lstinline^self->last_sender()^ is set.
This identifies the sender of the received message and is used implicitly by the functions \lstinline^reply^ and \lstinline^reply_tuple^.
Using \lstinline^reply(...)^ is \textbf{not} equal to \lstinline^send(self->last_sender(), ...)^.
The function \lstinline^send^ always uses asynchronous message passing, whereas \lstinline^reply^ will send a synchronous response message if the received message was a synchronous request (see Section \ref{Sec::Sync}).
%Note that you cannot reply more than once.
To delay a response, i.e., reply to a message after receiving another message, actors can use \lstinline^self->make_response_handle()^.
The functions \lstinline^reply_to^ and \lstinline^reply_tuple_to^ then can be used to reply the the original request, as shown in the example below.
This identifies the sender of the received message and is used implicitly to reply to the correct sender.
However, using \lstinline^send(self->last_sender(), ...)^ does \emph{not} reply to the message, i.e., synchronous messages will not recognize the message as response.
\begin{lstlisting}
void broker(const actor_ptr& master) {
void client(const actor_ptr& master) {
become (
on("foo", arg_match) >> [=](const std::string& request) {
auto hdl = make_response_handle();
sync_send(master, atom("bar"), request).then(
on("foo", arg_match) >> [=](const string& request) -> string {
return sync_send(master, atom("bar"), request).then(
on_arg_match >> [=](const std::string& response) {
reply_to(hdl, response);
return response;
}
);
}
......@@ -60,9 +54,6 @@ void broker(const actor_ptr& master) {
};
\end{lstlisting}
In any case, do never reply than more than once.
Additional (synchronous) response message will be ignored by the receiver.
\clearpage
\subsection{Chaining}
\label{Sec::Send::ChainedSend}
......
\section{Strongly Typed Actors}
Strongly typed actors provide a convenient way of defining type-safe messaging interfaces.
Unlike ``dynamic actors'', typed actors are not allowed to change their behavior at runtime, neither are typed actors allowed to use guard expressions.
When calling \lstinline^become^ in a strongly typed actor, the actor will be killed with exit reason \lstinline^unallowed_function_call^.
Typed actors use \lstinline^typed_actor_ptr<...>^ instead of \lstinline^actor_ptr^, whereas the template parameters hold the messaging interface.
For example, an actor responding to two integers with a dobule would use the type \lstinline^typed_actor_ptr<replies_to<int, int>::with<double>>^.
All functions for message passing, linking and monitoring are overloaded to accept both types of actors.
As of version 0.8, strongly typed actors cannot be published and do not support message priorities (those are planned feature for future releases).
\subsection{Spawning Typed Actors}
\label{sec:strong:spawn}
Actors are spawned using the function \lstinline^spawn_typed^.
The argument to this function call \emph{must} be a match expression as shown in the example below, because the runtime of libcppa needs to evaluate the signature of each message handler.
\begin{lstlisting}
auto p0 = spawn_typed(
on_arg_match >> [](int a, int b) {
return static_cast<double>(a) * b;
},
on_arg_match >> [](double a, double b) {
return make_cow_tuple(a * b, a / b);
}
);
// assign to identical type
using full_type = typed_actor_ptr<
replies_to<int, int>::with<double>,
replies_to<double, double>::with<double, double>
>;
full_type p1 = p0;
// assign to subtype
using subtype1 = typed_actor_ptr<
replies_to<int, int>::with<double>
>;
subtype1 p2 = p0;
// assign to another subtype
using subtype2 = typed_actor_ptr<
replies_to<double, double>::with<double, double>
>;
subtype2 p3 = p0;
\end{lstlisting}
\clearpage
\subsection{Class-based Typed Actors}
Typed actors can be implemented using a class by inheriting from \lstinline^typed_actor<...>^, whereas the template parameter pack denotes the messaging interface.
Derived classes have to implemented the pure virtual member function \lstinline^make_behavior^, as shown in the example below.
\begin{lstlisting}
// implementation
class typed_testee : public typed_actor<replies_to<int>::with<bool>> {
protected:
behavior_type make_behavior() final {
// returning a non-matching expression
// results in a compile-time error
return (
on_arg_match >> [](int value) {
return value == 42;
}
);
}
};
// instantiation
auto testee = spawn_typed<typed_testee>();
\end{lstlisting}
It is worth mentioning that \lstinline^typed_actor^ implements the member function \lstinline^init()^ using the \lstinline^final^ qualifier.
Hence, derived classes are not allowed to override \lstinline^init()^.
However, typed actors are allowed to override other member functions such as \lstinline^on_exit()^.
The return type of \lstinline^make_behavior^ is \lstinline^typed_behavior<...>^, which is aliased as \lstinline^behavior_type^ for convenience.
......@@ -35,7 +35,7 @@ When using synchronous messaging, \libcppa's runtime environment will send ...
\begin{itemize}
\item \lstinline^{'EXITED', uint32_t exit_reason}^ if the receiver is not alive
\item \lstinline^{'VOID'}^ if the receiver handled the message but did not call \lstinline^reply^
\item \lstinline^{'VOID'}^ if the receiver handled the message but did not respond to it
\item \lstinline^{'TIMEOUT'}^ if a message send by \lstinline^timed_sync_send^ timed out
\end{itemize}
......@@ -120,10 +120,11 @@ sync_send(testee, atom("get")).then(
);
\end{lstlisting}
\clearpage
\subsubsection{Continuations for Event-based Actors}
\libcppa supports continuations to enable chaining of send/receive statements.
The functions \lstinline^handle_response^ and \lstinline^message_future::then^ both return a helper object offering \lstinline^continue_with^, which takes a functor $f$ without arguments.
The functions \lstinline^handle_response^ and \lstinline^message_future::then^ both return a helper object offering the member function \lstinline^continue_with^, which takes a functor $f$ without arguments.
After receiving a message, $f$ is invoked if and only if the received messages was handled successfully, i.e., neither \lstinline^sync_failure^ nor \lstinline^sync_timeout^ occurred.
\begin{lstlisting}
......
......@@ -4,52 +4,27 @@
12pt,% % 12pt Schriftgröße
]{article}
%\usepackage{ifpdf}
%\usepackage[english]{babel}
% UTF8 input
\usepackage[utf8]{inputenc}
% grafics
%\usepackage{graphicx}
%\usepackage{picinpar}
% others
%\usepackage{parcolumns}
%\usepackage[font=footnotesize,format=plain,labelfont=bf]{caption}
%\usepackage{paralist}
% include required packages
\usepackage{url}
\usepackage{array}
%\usepackage{tocloft}
\usepackage{color}
\usepackage{listings}
\usepackage{xspace}
\usepackage{tabularx}
\usepackage{hyperref}
\usepackage{fancyhdr}
\usepackage{multicol}
% HTML compilation
\usepackage{hevea}
% font setup
\usepackage{cmbright}
\usepackage{txfonts}
\usepackage[scaled=.90]{helvet}
\usepackage{courier}
%\usepackage{natbib}
%\usepackage{lmodern}
%\usepackage{newcent}
%\renewcommand{\ttdefault}{lmtt}
%\usepackage{fontspec}
%\defaultfontfeatures{Mapping=tex-text} % For archaic input (e.g. convert -- to en-dash)
%\setmainfont{Times New Roman} % Computer Modern Unicode font
%\setmonofont{Anonymous Pro}
%\setsansfont{Times New Roman}
% par. settings
\parindent 0pt
......@@ -70,24 +45,23 @@
%BEGIN LATEX
\texttt{\huge{\textbf{libcppa}}}\\~\\A C++ library for actor programming\\~\\~\\~\\%
%END LATEX
User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
User Manual\\\normalsize{\texttt{libcppa} version 0.8.1}\vfill}
\author{Dominik Charousset}
\date{\today}
% page setup
\setlength{\voffset}{-0.5in}
\setlength{\hoffset}{-0.5in}
\addtolength{\textwidth}{1in}
\addtolength{\textheight}{1.5in}
\setlength{\headheight}{15pt}
% some HEVEA / HTML style sheets
\newstyle{body}{width:600px;margin:auto;padding-top:20px;text-align: justify;}
% more compact itemize
\newenvironment{itemize*}%
{\begin{itemize}%
\setlength{\itemsep}{0pt}%
......@@ -95,8 +69,8 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
{\end{itemize}}
\begin{document}
%\fontfamily{ptm}\selectfont
% fancy header setup
%BEGIN LATEX
\fancypagestyle{plain}{%
\fancyhf{} % clear all header and footer fields
......@@ -104,12 +78,14 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
\renewcommand{\footrulewidth}{0pt}
}
%END LATEX
\maketitle
\clearpage
\pagestyle{empty}
\tableofcontents
\clearpage
% fancy header setup for page 1
%BEGIN LATEX
\setcounter{page}{1}
\pagenumbering{arabic}
......@@ -121,12 +97,10 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
}
%END LATEX
%{\parskip=0mm \tableofcontents}
% basic setup for listings
% code listings setup
\lstset{%
language=C++,%
morekeywords={constexpr,nullptr,size_t,uint32_t,assert},%
morekeywords={constexpr,nullptr,size_t,uint32_t,assert,override,final},%
basicstyle=\ttfamily\small,%
sensitive=true,%
keywordstyle=\color{blue},%
......@@ -137,6 +111,7 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
numberstyle=\footnotesize%
}
% content
\include{FirstSteps}
\include{CopyOnWriteTuples}
\include{PatternMatching}
......@@ -147,21 +122,13 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.8}\vfill}
\include{ActorManagement}
\include{SpawningActors}
\include{MessagePriorities}
\include{StronglyTypedActors}
\include{NetworkTransparency}
\include{GroupCommunication}
\include{ActorCompanions}
\include{TypeSystem}
\include{BlockingAPI}
%\include{OpenCL}
\include{StronglyTypedActors}
\include{CommonPitfalls}
\include{Appendix}
%\clearpage
%\bibliographystyle{dinat}
%\bibliographystyle{apalike}
%\bibliography{/bib/programming,/bib/rfcs}
%\clearpage
%{\parskip=0mm \listoffigures}
%\clearpage
\end{document}
......@@ -193,6 +193,7 @@ void actor::cleanup(std::uint32_t reason) {
<< ", class = " << detail::demangle(typeid(*this)));
// send exit messages
auto msg = make_any_tuple(atom("EXIT"), reason);
CPPA_LOGM_DEBUG("cppa::actor", "send EXIT to " << mlinks.size() << " links");
for (actor_ptr& aptr : mlinks) {
message_header hdr{this, aptr, message_priority::high};
hdr.deliver(msg);
......
......@@ -39,9 +39,11 @@ class continuation_decorator : public detail::behavior_impl {
public:
typedef behavior::continuation_fun continuation_fun;
typedef typename behavior_impl::pointer pointer;
continuation_decorator(const partial_function& fun, pointer ptr)
continuation_decorator(continuation_fun fun, pointer ptr)
: super(ptr->timeout()), m_fun(fun), m_decorated(std::move(ptr)) {
CPPA_REQUIRE(m_decorated != nullptr);
}
......@@ -73,15 +75,15 @@ class continuation_decorator : public detail::behavior_impl {
private:
partial_function m_fun;
continuation_fun m_fun;
pointer m_decorated;
};
behavior::behavior(const partial_function& fun) : m_impl(fun.m_impl) { }
behavior behavior::add_continuation(const partial_function& fun) {
return {new continuation_decorator(fun, m_impl)};
behavior behavior::add_continuation(continuation_fun fun) {
return {new continuation_decorator(std::move(fun), m_impl)};
}
......
......@@ -43,6 +43,7 @@
#include "cppa/type_lookup_table.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/detail/ieee_754.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
using namespace std;
......@@ -68,15 +69,35 @@ inline void range_check(pointer begin, pointer end, size_t read_size) {
}
}
pointer read_range(pointer begin, pointer end, string& storage);
template<typename T>
pointer read_range(pointer begin, pointer end, T& storage,
// typename enable_if<is_integral<T>::value>::type* = 0) {
typename enable_if<is_arithmetic<T>::value>::type* = 0) {
typename enable_if<is_integral<T>::value>::type* = 0) {
range_check(begin, end, sizeof(T));
memcpy(&storage, begin, sizeof(T));
return advanced(begin, sizeof(T));
}
template<typename T>
pointer read_range(pointer begin, pointer end, T& storage,
typename enable_if<is_floating_point<T>::value>::type* = 0) {
typename detail::ieee_754_trait<T>::packed_type tmp;
auto result = read_range(begin, end, tmp);
storage = detail::unpack754(tmp);
return result;
}
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks)
pointer read_range(pointer begin, pointer end, long double& storage) {
std::string tmp;
auto result = read_range(begin, end, tmp);
std::istringstream iss{std::move(tmp)};
iss >> storage;
return result;
}
pointer read_range(pointer begin, pointer end, string& storage) {
uint32_t str_size;
begin = read_range(begin, end, str_size);
......
......@@ -30,6 +30,7 @@
#include <limits>
#include <string>
#include <iomanip>
#include <cstdint>
#include <sstream>
#include <cstring>
......@@ -39,15 +40,14 @@
#include "cppa/binary_serializer.hpp"
#include "cppa/type_lookup_table.hpp"
#include "cppa/detail/ieee_754.hpp"
using std::enable_if;
namespace cppa {
namespace {
constexpr size_t chunk_size = 512;
constexpr size_t ui32_size = sizeof(std::uint32_t);
using util::grow_if_needed;
class binary_writer {
......@@ -69,10 +69,25 @@ class binary_writer {
template<typename T>
void operator()(const T& value,
typename enable_if<std::is_arithmetic<T>::value>::type* = 0) {
typename enable_if<std::is_integral<T>::value>::type* = 0) {
write_int(m_sink, value);
}
template<typename T>
void operator()(const T& value,
typename enable_if<std::is_floating_point<T>::value>::type* = 0) {
auto tmp = detail::pack754(value);
write_int(m_sink, tmp);
}
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks)
void operator()(const long double& v) {
std::ostringstream oss;
oss << std::setprecision(std::numeric_limits<long double>::digits) << v;
write_string(m_sink, oss.str());
}
void operator()(const atom_value& val) {
(*this)(static_cast<uint64_t>(val));
}
......
......@@ -113,8 +113,8 @@ class broker::servant : public continuable {
: super{std::forward<Ts>(args)...}, m_disconnected{false}
, m_parent{move(parent)} { }
void io_failed() override {
disconnect();
void io_failed(event_bitmask mask) override {
if (mask == event::read) disconnect();
}
void dispose() override {
......@@ -422,7 +422,7 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
mm->continue_reader(kvp.second.get());
});
}
return move(ptr);
return ptr;
}
broker_ptr broker::from_impl(std::function<void (broker*)> fun,
......
......@@ -77,9 +77,11 @@ default_peer::default_peer(default_protocol* parent,
m_meta_msg = uniform_typeid<any_tuple>();
}
void default_peer::io_failed() {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr"));
if (m_node) {
void default_peer::io_failed(event_bitmask mask) {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr")
<< " mask = " << mask);
// make sure this code is executed only once by filtering for read failure
if (mask == event::read && m_node) {
// kill all proxies
auto& children = m_parent->addressing()->proxies(*m_node);
for (auto& kvp : children) {
......
......@@ -83,7 +83,7 @@ continue_reading_result default_peer_acceptor::continue_reading() {
}
}
void default_peer_acceptor::io_failed() {
void default_peer_acceptor::io_failed(event_bitmask) {
CPPA_LOG_INFO("removed default_peer_acceptor "
<< this << " due to an IO failure");
}
......
......@@ -158,10 +158,10 @@ void local_actor::reply_message(any_tuple&& what) {
}
}
void local_actor::forward_message(const actor_ptr& new_receiver) {
if (new_receiver == nullptr) return;
void local_actor::forward_message(const actor_ptr& dest, message_priority p) {
if (dest == nullptr) return;
auto& id = m_current_node->mid;
new_receiver->enqueue({last_sender(), new_receiver, id}, m_current_node->msg);
dest->enqueue({last_sender(), dest, id, p}, m_current_node->msg);
// treat this message as asynchronous message from now on
id = message_id{};
}
......@@ -170,7 +170,7 @@ response_handle local_actor::make_response_handle() {
auto n = m_current_node;
response_handle result{this, n->sender, n->mid.response_id()};
n->mid.mark_as_answered();
return std::move(result);
return result;
}
void local_actor::exec_behavior_stack() {
......
......@@ -232,7 +232,9 @@ class middleman_overseer : public continuable {
return read_continue_later;
}
void io_failed() { CPPA_CRITICAL("IO on pipe failed"); }
void io_failed(event_bitmask) override {
CPPA_CRITICAL("IO on pipe failed");
}
private:
......@@ -305,8 +307,10 @@ void middleman_loop(middleman_impl* impl) {
case event::write: {
CPPA_LOGF_DEBUG("handle event::write for " << io);
switch (io->continue_writing()) {
case read_failure:
io->io_failed(event::write);
// fall through
case write_closed:
case write_failure:
impl->stop_writer(io);
CPPA_LOGF_DEBUG("writer removed because of error");
break;
......@@ -322,8 +326,10 @@ void middleman_loop(middleman_impl* impl) {
case event::read: {
CPPA_LOGF_DEBUG("handle event::read for " << io);
switch (io->continue_reading()) {
case read_closed:
case read_failure:
io->io_failed(event::read);
// fall through
case read_closed:
impl->stop_reader(io);
CPPA_LOGF_DEBUG("remove peer");
break;
......@@ -333,7 +339,8 @@ void middleman_loop(middleman_impl* impl) {
}
case event::error: {
CPPA_LOGF_DEBUG("event::error; remove peer " << io);
io->io_failed();
io->io_failed(event::write);
io->io_failed(event::read);
impl->stop_reader(io);
impl->stop_writer(io);
}
......@@ -354,8 +361,10 @@ void middleman_loop(middleman_impl* impl) {
switch (mask) {
case event::write:
switch (io->continue_writing()) {
case write_closed:
case write_failure:
io->io_failed(event::write);
// fall through
case write_closed:
case write_done:
handler->erase_later(io, event::write);
break;
......@@ -363,7 +372,8 @@ void middleman_loop(middleman_impl* impl) {
}
break;
case event::error:
io->io_failed();
io->io_failed(event::write);
io->io_failed(event::read);
handler->erase_later(io, event::both);
break;
default:
......
......@@ -40,3 +40,19 @@ partial_function::partial_function(impl_ptr ptr) : m_impl(std::move(ptr)) { }
void detail::behavior_impl::handle_timeout() { }
} // namespace cppa
namespace cppa { namespace detail {
behavior_impl_ptr combine(behavior_impl_ptr lhs, const partial_function& rhs) {
return lhs->or_else(rhs.as_behavior_impl());
}
behavior_impl_ptr combine(const partial_function& lhs, behavior_impl_ptr rhs) {
return lhs.as_behavior_impl()->or_else(rhs);
}
behavior_impl_ptr extract(const partial_function& arg) {
return arg.as_behavior_impl();
}
} } // namespace cppa::detail
......@@ -66,17 +66,9 @@ class delayed_msg {
public:
delayed_msg(const channel_ptr& arg0,
const actor_ptr& arg1,
message_id,
any_tuple&& arg3)
: ptr_a(arg0), from(arg1), msg(move(arg3)) { }
delayed_msg(const actor_ptr& arg0,
const actor_ptr& arg1,
message_id arg2,
any_tuple&& arg3)
: ptr_b(arg0), from(arg1), id(arg2), msg(move(arg3)) { }
delayed_msg(message_header&& arg1,
any_tuple&& arg2)
: hdr(move(arg1)), msg(move(arg2)) { }
delayed_msg(delayed_msg&&) = default;
delayed_msg(const delayed_msg&) = default;
......@@ -84,17 +76,12 @@ class delayed_msg {
delayed_msg& operator=(const delayed_msg&) = default;
inline void eval() {
CPPA_REQUIRE(ptr_a || ptr_b);
if (ptr_a) ptr_a->enqueue(from, move(msg));
else ptr_b->enqueue({from, id}, move(msg));
hdr.deliver(std::move(msg));
}
private:
channel_ptr ptr_a;
actor_ptr ptr_b;
actor_ptr from;
message_id id;
message_header hdr;
any_tuple msg;
};
......@@ -140,16 +127,14 @@ class scheduler_helper {
};
template<class Map, class T>
template<class Map>
inline void insert_dmsg(Map& storage,
const util::duration& d,
const T& to,
const actor_ptr& sender,
any_tuple&& tup,
message_id id = message_id{}) {
message_header&& hdr,
any_tuple&& tup ) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{to, sender, id, move(tup)};
delayed_msg dmsg{move(hdr), move(tup)};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
......@@ -163,15 +148,9 @@ void scheduler_helper::timer_loop(scheduler_helper::ptr_type m_self) {
// message handling rules
auto mfun = (
on(atom("SEND"), arg_match) >> [&](const util::duration& d,
const channel_ptr& ptr,
message_header& hdr,
any_tuple& tup) {
insert_dmsg(messages, d, ptr, msg_ptr->sender, move(tup));
},
on(atom("REPLY"), arg_match) >> [&](const util::duration& d,
const actor_ptr& ptr,
message_id id,
any_tuple& tup) {
insert_dmsg(messages, d, ptr, msg_ptr->sender, move(tup), id);
insert_dmsg(messages, d, move(hdr), move(tup));
},
on(atom("DIE")) >> [&] {
done = true;
......
......@@ -34,10 +34,10 @@
namespace cppa {
message_future sync_send_tuple(actor_ptr whom, any_tuple what) {
if (!whom) throw std::invalid_argument("whom == nullptr");
message_future sync_send_tuple(actor_destination dest, any_tuple what) {
if (!dest.receiver) throw std::invalid_argument("whom == nullptr");
auto req = self->new_request_id();
message_header hdr{self, std::move(whom), req};
message_header hdr{self, std::move(dest.receiver), req, dest.priority};
if (self->chaining_enabled()) {
if (hdr.receiver->chained_enqueue(hdr, std::move(what))) {
self->chained_actor(hdr.receiver.downcast<actor>());
......@@ -47,27 +47,30 @@ message_future sync_send_tuple(actor_ptr whom, any_tuple what) {
return req.response_id();
}
void delayed_send_tuple(const channel_ptr& to,
const util::duration& rel_time,
void delayed_send_tuple(channel_destination dest,
const util::duration& rtime,
any_tuple data) {
if (to) get_scheduler()->delayed_send(to, rel_time, std::move(data));
if (dest.receiver) {
message_header hdr{self, std::move(dest.receiver), dest.priority};
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(data));
}
}
void delayed_reply_tuple(const util::duration& rel_time,
actor_ptr whom,
message_id mid,
any_tuple data) {
if (whom) get_scheduler()->delayed_reply(whom,
rel_time,
mid,
std::move(data));
message_future timed_sync_send_tuple(actor_destination dest,
const util::duration& rtime,
any_tuple what) {
auto mf = sync_send_tuple(std::move(dest), std::move(what));
message_header hdr{self, self, mf.id()};
auto tmp = make_any_tuple(atom("TIMEOUT"));
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(tmp));
return mf;
}
void delayed_reply_tuple(const util::duration& rel_time,
void delayed_reply_tuple(const util::duration& rtime,
message_id mid,
any_tuple data) {
delayed_reply_tuple(rel_time, self->last_sender(), mid, std::move(data));
message_header hdr{self, self->last_sender(), mid};
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(data));
}
void delayed_reply_tuple(const util::duration& rel_time, any_tuple data) {
......
......@@ -207,7 +207,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_pt
exec_as_thread(is_hidden, p, [p] {
p->run_detached();
});
return std::move(p);
return p;
}
p->attach_to_scheduler(this, is_hidden);
if (p->has_behavior() || p->impl_type() == default_event_based_impl) {
......@@ -216,7 +216,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_pt
if (p->impl_type() != event_based_impl) m_queue.push_back(p.get());
}
else p->on_exit();
return std::move(p);
return p;
}
local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
......@@ -274,7 +274,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
exec(os, p);
}
CPPA_REQUIRE(result != nullptr);
return std::move(result);
return result;
}
} } // namespace cppa::detail
......@@ -785,7 +785,7 @@ class utim_impl : public uniform_type_info_map {
res.reserve(m_builtin_types.size() + m_user_types.size());
res.insert(res.end(), m_builtin_types.begin(), m_builtin_types.end());
res.insert(res.end(), m_user_types.begin(), m_user_types.end());
return std::move(res);
return res;
}
pointer insert(std::unique_ptr<uniform_type_info> uti) {
......
......@@ -24,6 +24,7 @@ add_unit_test(primitive_variant)
add_unit_test(yield_interface)
add_unit_test(tuple)
add_unit_test(spawn ping_pong.cpp)
add_unit_test(typed_spawn)
add_unit_test(local_group)
add_unit_test(sync_send)
add_unit_test(remote_actor ping_pong.cpp)
......
......@@ -24,7 +24,7 @@ behavior ping_behavior(size_t num_pings) {
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);
send_exit(self->last_sender(), exit_reason::user_shutdown);
self->quit();
}
return {atom("ping"), value};
......@@ -32,7 +32,7 @@ behavior ping_behavior(size_t num_pings) {
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
<< to_string(self->last_dequeued()));
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
}
);
}
......@@ -46,7 +46,7 @@ behavior pong_behavior() {
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
<< to_string(self->last_dequeued()));
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
}
);
}
......
......@@ -19,7 +19,7 @@ void cppa_inc_error_count() {
string cppa_fill4(int value) {
string result = to_string(value);
while (result.size() < 4) result.insert(result.begin(), '0');
return move(result);
return result;
}
const char* cppa_strip_path(const char* fname) {
......
......@@ -31,7 +31,7 @@ void testee(int current_value, int final_result) {
},
after(std::chrono::seconds(2)) >> [] {
CPPA_UNEXPECTED_TOUT();
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
}
);
}
......
......@@ -29,8 +29,6 @@ int main() {
CPPA_TEST(test_metaprogramming);
CPPA_PRINT("test type_list");
typedef type_list<int, float, std::string> l1;
typedef typename tl_reverse<l1>::type r1;
......@@ -52,8 +50,6 @@ int main() {
CPPA_CHECK_EQUAL((util::tl_count<l1, is_int>::value), 1);
CPPA_CHECK_EQUAL((util::tl_count<l2, is_int>::value), 2);
CPPA_PRINT("test int_list");
typedef int_list<0, 1, 2, 3, 4, 5> il0;
typedef int_list<4, 5> il1;
typedef typename il_right<il0, 2>::type il2;
......@@ -67,6 +63,7 @@ int main() {
CPPA_CHECK((tl_is_strict_subset<list_a, list_b>::value));
CPPA_CHECK(!(tl_is_strict_subset<list_b, list_a>::value));
CPPA_CHECK((tl_is_strict_subset<list_a, list_a>::value));
CPPA_CHECK((tl_is_strict_subset<list_b, list_b>::value));
}
return CPPA_TEST_RESULT();
......
......@@ -51,6 +51,7 @@
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/detail/ieee_754.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/type_to_ptype.hpp"
#include "cppa/detail/ptype_to_type.hpp"
......@@ -138,9 +139,26 @@ struct raw_struct_type_info : util::abstract_uniform_type_info<raw_struct> {
}
};
void test_ieee_754() {
// check float packing
float f1 = 3.1415925; // float value
auto p1 = cppa::detail::pack754(f1); // packet value
CPPA_CHECK_EQUAL(p1, 0x40490FDA);
auto u1 = cppa::detail::unpack754(p1); // unpacked value
CPPA_CHECK_EQUAL(f1, u1);
// check double packing
double f2 = 3.14159265358979311600; // float value
auto p2 = cppa::detail::pack754(f2); // packet value
CPPA_CHECK_EQUAL(p2, 0x400921FB54442D18);
auto u2 = cppa::detail::unpack754(p2); // unpacked value
CPPA_CHECK_EQUAL(f2, u2);
}
int main() {
CPPA_TEST(test_serialization);
test_ieee_754();
typedef std::integral_constant<int, detail::impl_id<strmap>()> token;
CPPA_CHECK_EQUAL(util::is_iterable<strmap>::value, true);
CPPA_CHECK_EQUAL(detail::is_stl_compliant_list<vector<int>>::value, true);
......
This diff is collapsed.
......@@ -104,12 +104,13 @@ struct D : popular_actor {
void init() {
become (
others() >> [=] {
/*
response_handle handle = make_response_handle();
sync_send_tuple(buddy(), last_dequeued()).then([=] {
reply_to(handle, last_dequeued());
quit();
});
/*/
//*/
return sync_send_tuple(buddy(), last_dequeued()).then([=]() -> any_tuple {
quit();
return last_dequeued();
......@@ -138,7 +139,7 @@ struct D : popular_actor {
struct server : event_based_actor {
void init() {
auto die = [=] { quit(exit_reason::user_defined); };
auto die = [=] { quit(exit_reason::user_shutdown); };
become (
on(atom("idle")) >> [=] {
auto worker = last_sender();
......@@ -201,10 +202,10 @@ int main() {
);
CPPA_CHECK_EQUAL(sync_failure_called, true);
CPPA_CHECK_EQUAL(int_handler_called, false);
self->quit(exit_reason::user_defined);
self->quit(exit_reason::user_shutdown);
});
receive (
on(atom("DOWN"), exit_reason::user_defined) >> CPPA_CHECKPOINT_CB(),
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
auto mirror = spawn<sync_mirror>();
......@@ -214,7 +215,7 @@ int main() {
.continue_with([&] { continuation_called = true; });
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(continuation_called, true);
send_exit(mirror, exit_reason::user_defined);
send_exit(mirror, exit_reason::user_shutdown);
await_all_others_done();
CPPA_CHECKPOINT();
auto await_success_message = [&] {
......@@ -245,7 +246,11 @@ int main() {
int i = 0;
receive_for(i, 3) (
on(atom("DOWN"), exit_reason::normal) >> CPPA_CHECKPOINT_CB(),
on(atom("NoWay")) >> CPPA_CHECKPOINT_CB(),
on(atom("NoWay")) >> [] {
CPPA_CHECKPOINT();
CPPA_PRINT("trigger \"actor did not reply to a "
"synchronous request message\"");
},
others() >> CPPA_UNEXPECTED_MSG_CB(),
after(std::chrono::seconds(0)) >> CPPA_UNEXPECTED_TOUT_CB()
);
......@@ -270,7 +275,7 @@ int main() {
sync_send(c, atom("gogo")).then(CPPA_CHECKPOINT_CB())
.continue_with(CPPA_CHECKPOINT_CB());
self->exec_behavior_stack();
send_exit(c, exit_reason::user_defined);
send_exit(c, exit_reason::user_shutdown);
await_all_others_done();
CPPA_CHECKPOINT();
......@@ -300,11 +305,11 @@ int main() {
others() >> CPPA_UNEXPECTED_MSG_CB()
);
send(s, "Ever danced with the devil in the pale moonlight?");
// response: {'EXIT', exit_reason::user_defined}
// response: {'EXIT', exit_reason::user_shutdown}
receive_loop(others() >> CPPA_UNEXPECTED_MSG_CB());
});
receive (
on(atom("DOWN"), exit_reason::user_defined) >> CPPA_CHECKPOINT_CB(),
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
await_all_others_done();
......
......@@ -407,8 +407,24 @@ void check_wildcards() {
}
}
void check_move_optional() {
CPPA_PRINT(__func__);
optional<expensive_copy_struct> opt{expensive_copy_struct{}};
opt->value = 23;
auto opt2 = std::move(opt);
auto move_fun = [](expensive_copy_struct& value) -> optional<expensive_copy_struct> {
return std::move(value);
};
auto opt3 = move_fun(*opt2);
CPPA_CHECK(opt3.valid());
CPPA_CHECK_EQUAL(opt->value, 23);
CPPA_CHECK_EQUAL(s_expensive_copies.load(), 0);
}
void check_move_ops() {
check_move_optional();
CPPA_PRINT(__func__);
CPPA_CHECK_EQUAL(s_expensive_copies.load(), 0);
send(spawn<dummy_receiver>(), expensive_copy_struct());
receive (
on_arg_match >> [&](expensive_copy_struct& ecs) {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 "cppa/cppa.hpp"
#include "test.hpp"
using namespace std;
using namespace cppa;
struct my_request { int a; int b; };
bool operator==(const my_request& lhs, const my_request& rhs) {
return lhs.a == rhs.a && lhs.b == rhs.b;
}
typed_actor_ptr<replies_to<my_request>::with<bool>>
spawn_typed_server() {
return spawn_typed(
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
}
);
}
class typed_testee : public typed_actor<replies_to<my_request>::with<bool>> {
protected:
behavior_type make_behavior() override {
return (
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
}
);
}
};
int main() {
CPPA_TEST(test_typed_spawn);
announce<my_request>(&my_request::a, &my_request::b);
auto sptr = spawn_typed_server();
sync_send(sptr, my_request{2, 2}).await(
[](bool value) {
CPPA_CHECK_EQUAL(value, true);
}
);
send_exit(sptr, exit_reason::user_shutdown);
sptr = spawn_typed<typed_testee>();
sync_send(sptr, my_request{2, 2}).await(
[](bool value) {
CPPA_CHECK_EQUAL(value, true);
}
);
send_exit(sptr, exit_reason::user_shutdown);
auto ptr0 = spawn_typed(
on_arg_match >> [](double d) {
return d * d;
},
on_arg_match >> [](float f) {
return f / 2.0f;
}
);
CPPA_CHECK((std::is_same<
decltype(ptr0),
typed_actor_ptr<
replies_to<double>::with<double>,
replies_to<float>::with<float>
>
>::value));
typed_actor_ptr<replies_to<double>::with<double>> ptr0_double = ptr0;
typed_actor_ptr<replies_to<float>::with<float>> ptr0_float = ptr0;
auto ptr = spawn_typed(
on<int>() >> [] { return "wtf"; },
on<string>() >> [] { return 42; },
on<float>() >> [] { return make_cow_tuple(1, 2, 3); },
on<double>() >> [=](double d) {
return sync_send(ptr0_double, d).then(
[](double res) { return res + res; }
);
}
);
// check async messages
send(ptr0_float, 4.0f);
receive(
on_arg_match >> [](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
}
);
// check sync messages
sync_send(ptr0_float, 4.0f).await(
[](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
}
);
sync_send(ptr, 10.0).await(
[](double d) {
CPPA_CHECK_EQUAL(d, 200.0);
}
);
sync_send(ptr, 42).await(
[](const std::string& str) {
CPPA_CHECK_EQUAL(str, "wtf");
}
);
sync_send(ptr, 1.2f).await(
[](int a, int b, int c) {
CPPA_CHECK_EQUAL(a, 1);
CPPA_CHECK_EQUAL(b, 2);
CPPA_CHECK_EQUAL(c, 3);
}
);
sync_send(ptr, 1.2f).await(
[](int b, int c) {
CPPA_CHECK_EQUAL(b, 2);
CPPA_CHECK_EQUAL(c, 3);
}
);
sync_send(ptr, 1.2f).await(
[](int c) {
CPPA_CHECK_EQUAL(c, 3);
}
);
sync_send(ptr, 1.2f).await(
[] { CPPA_CHECKPOINT(); }
);
spawn([=] {
sync_send(ptr, 2.3f).then(
[] (int c) {
CPPA_CHECK_EQUAL(c, 3);
return "hello continuation";
}
).continue_with(
[] (const string& str) {
CPPA_CHECK_EQUAL(str, "hello continuation");
return 4.2;
}
).continue_with(
[=] (double d) {
CPPA_CHECK_EQUAL(d, 4.2);
send_exit(ptr0, exit_reason::user_shutdown);
send_exit(ptr, exit_reason::user_shutdown);
self->quit();
}
);
});
await_all_others_done();
CPPA_CHECKPOINT();
shutdown();
return CPPA_TEST_RESULT();
}
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