Commit 776fa09f authored by Dominik Charousset's avatar Dominik Charousset

bugfixes for synchronous messages

parent 804047b1
......@@ -338,3 +338,4 @@ src/untyped_actor.cpp
src/functor_based_blocking_actor.cpp
src/blocking_untyped_actor.cpp
cppa/policy/policies.hpp
cppa/detail/response_future_util.hpp
......@@ -31,6 +31,7 @@
#ifndef CPPA_ACTOR_HPP
#define CPPA_ACTOR_HPP
#include <cstddef>
#include <cstdint>
#include <type_traits>
......@@ -43,6 +44,10 @@
namespace cppa {
class actor_proxy;
class untyped_actor;
class blocking_untyped_actor;
namespace detail { class raw_access; }
struct invalid_actor_t { constexpr invalid_actor_t() { } };
......@@ -61,16 +66,28 @@ class actor : util::comparable<actor> {
actor() = default;
template<typename T>
actor(intrusive_ptr<T> ptr, typename std::enable_if<std::is_base_of<abstract_actor, T>::value>::type* = 0) : m_ops(ptr) { }
actor(intrusive_ptr<T> ptr,
typename std::enable_if<
std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<untyped_actor, T>::value
|| std::is_base_of<blocking_untyped_actor, T>::value
>::type* = 0)
: m_ops(ptr) { }
actor(abstract_actor*);
actor(const std::nullptr_t&);
explicit inline operator bool() const;
actor(actor_proxy*);
inline bool operator!() const;
actor(untyped_actor*);
actor(blocking_untyped_actor*);
actor(const invalid_actor_t&);
explicit inline operator bool() const;
inline bool operator!() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
inline common_actor_ops* operator->() const {
......@@ -84,6 +101,8 @@ class actor : util::comparable<actor> {
private:
actor(abstract_actor*);
common_actor_ops m_ops;
};
......
......@@ -45,6 +45,8 @@ namespace cppa {
class actor;
class local_actor;
class actor_namespace;
namespace detail { class raw_access; }
struct invalid_actor_addr_t { constexpr invalid_actor_addr_t() { } };
......@@ -53,7 +55,8 @@ constexpr invalid_actor_addr_t invalid_actor_addr = invalid_actor_addr_t{};
class actor_addr : util::comparable<actor_addr>
, util::comparable<actor_addr, actor>
, util::comparable<actor_addr, local_actor*> {
, util::comparable<actor_addr, abstract_actor*>
, util::comparable<actor_addr, abstract_actor_ptr> {
friend class abstract_actor;
friend class detail::raw_access;
......@@ -77,7 +80,11 @@ class actor_addr : util::comparable<actor_addr>
intptr_t compare(const actor& other) const;
intptr_t compare(const actor_addr& other) const;
intptr_t compare(const local_actor* other) const;
intptr_t compare(const abstract_actor* other) const;
inline intptr_t compare(const abstract_actor_ptr& other) const {
return compare(other.get());
}
inline common_actor_ops* operator->() const {
// this const cast is safe, because common_actor_ops cannot be
......
......@@ -38,6 +38,8 @@
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/detail/response_future_util.hpp"
namespace cppa {
/**
......@@ -76,7 +78,7 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
await(behavior{(on_arg_match >> std::move(fs))...});
await(detail::fs2bhvr(m_self, fs...));
}
/**
......
......@@ -590,12 +590,12 @@ struct actor_ostream {
constexpr actor_ostream() { }
inline const actor_ostream& write(std::string arg) const {
send_as(nullptr, get_scheduler()->printer(), atom("add"), move(arg));
send_as(invalid_actor, get_scheduler()->printer(), atom("add"), move(arg));
return *this;
}
inline const actor_ostream& flush() const {
send_as(nullptr, get_scheduler()->printer(), atom("flush"));
send_as(invalid_actor, get_scheduler()->printer(), atom("flush"));
return *this;
}
......
......@@ -172,15 +172,14 @@ class proper_actor : public proper_actor_base<Base,
: super(std::forward<Ts>(args)...), m_has_timeout(false)
, m_timeout_id(0) { }
inline void launch() {
inline void launch(bool is_hidden) {
CPPA_LOG_TRACE("");
auto bhvr = this->make_behavior();
CPPA_LOG_DEBUG_IF(!bhvr, "make_behavior() returned an empty behavior");
if (bhvr) this->become(std::move(bhvr));
CPPA_LOG_WARNING_IF(this->bhvr_stack().empty(),
"actor did not set a behavior");
if (!this->bhvr_stack().empty()) {
this->scheduling_policy().launch(this);
this->scheduling_policy().launch(this, is_hidden);
}
}
......@@ -299,8 +298,8 @@ class proper_actor<Base, Policies,true> : public proper_actor_base<Base,
this->resume_policy().await_ready(this);
}
inline void launch() {
this->scheduling_policy().launch(this);
inline void launch(bool is_hidden) {
this->scheduling_policy().launch(this, is_hidden);
}
// implement blocking_untyped_actor::dequeue_response
......
......@@ -55,6 +55,14 @@ class raw_access {
return hdl.m_ptr.get();
}
static actor unsafe_cast(abstract_actor* ptr) {
return {ptr};
}
static actor unsafe_cast(const actor_addr& hdl) {
return {get(hdl)};
}
};
// utility function to get raw access + cast to a related type in one call
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_DETAIL_RESPONSE_FUTURE_UTIL_HPP
#define CPPA_DETAIL_RESPONSE_FUTURE_UTIL_HPP
#include "cppa/on.hpp"
#include "cppa/match_hint.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
namespace detail {
template<typename Actor, typename... Fs>
behavior fs2bhvr(Actor* self, Fs... fs) {
auto handle_sync_timeout = [self]() -> match_hint {
self->handle_sync_timeout();
return match_hint::skip;
};
return behavior{
on(atom("TIMEOUT")) >> handle_sync_timeout,
on(atom("VOID")) >> skip_message,
on(atom("EXITED")) >> skip_message,
(on(any_vals, arg_match) >> std::move(fs))...
};
}
} // namespace detail
} // namespace cppa
#endif // CPPA_DETAIL_RESPONSE_FUTURE_UTIL_HPP
......@@ -74,7 +74,7 @@ class remote_actor_proxy : public actor_proxy {
public:
remote_actor_proxy(actor_id mid,
const node_id_ptr& pinfo,
node_id_ptr pinfo,
middleman* parent);
void enqueue(const message_header& hdr, any_tuple msg) override;
......@@ -93,10 +93,6 @@ class remote_actor_proxy : public actor_proxy {
void deliver(const message_header& hdr, any_tuple msg) override;
inline const node_id_ptr& process_info() const {
return m_pinf;
}
protected:
~remote_actor_proxy();
......@@ -106,7 +102,6 @@ class remote_actor_proxy : public actor_proxy {
void forward_msg(const message_header& hdr, any_tuple msg);
middleman* m_parent;
node_id_ptr m_pinf;
intrusive::single_reader_queue<sync_request_info, detail::disposer> m_pending_requests;
};
......
......@@ -40,6 +40,7 @@
#include "cppa/extend.hpp"
#include "cppa/channel.hpp"
#include "cppa/behavior.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/match_expr.hpp"
......@@ -73,12 +74,21 @@ namespace util {
struct fiber;
} // namespace util
// prototype definitions of the spawn function famility;
// implemented in spawn.hpp (this header is included there)
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args);
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args);
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr&, Ts&&... args);
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr&, Ts&&... args);
/**
* @brief Base class for local running Actors.
* @extends actor
......@@ -118,6 +128,18 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
return eval_opts(Options, std::move(res));
}
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
auto res = cppa::spawn_in_group<make_unbound(Options)>(grp, std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
template<class Impl, spawn_options Options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
auto res = cppa::spawn_in_group<Impl, make_unbound(Options)>(grp, std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
/**
* @brief Sends @p what to the receiver specified in @p dest.
*/
......@@ -331,7 +353,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
inline bool chaining_enabled();
message_id send_timed_sync_message(message_priority mp,
message_id timed_sync_send_tuple_impl(message_priority mp,
const actor& whom,
const util::duration& rel_time,
any_tuple&& what);
......
......@@ -104,17 +104,17 @@ class no_scheduling {
}
template<class Actor>
void launch(Actor* self) {
void launch(Actor* self, bool is_hidden) {
CPPA_PUSH_AID(self->id());
CPPA_LOG_TRACE(CPPA_ARG(self));
CPPA_LOG_TRACE(CPPA_ARG(self) << ", " << CPPA_ARG(is_hidden));
CPPA_REQUIRE(self != nullptr);
get_actor_registry()->inc_running();
if (!is_hidden) get_actor_registry()->inc_running();
intrusive_ptr<Actor> mself{self};
std::thread([=] {
CPPA_PUSH_AID(mself->id());
CPPA_LOG_TRACE("");
auto guard = util::make_scope_guard([] {
get_actor_registry()->dec_running();
auto guard = util::make_scope_guard([is_hidden] {
if (!is_hidden) get_actor_registry()->dec_running();
});
util::fiber fself;
for (;;) {
......
......@@ -36,6 +36,7 @@
#include "cppa/policy.hpp"
#include "cppa/logging.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/spawn_options.hpp"
......@@ -51,8 +52,8 @@ namespace cppa {
* @{
*/
template<class Impl, spawn_options Options, typename... Ts>
actor spawn_impl(Ts&&... args) {
template<class Impl, spawn_options Options, typename BeforeLaunch, typename... Ts>
actor spawn_impl(BeforeLaunch before_launch_fun, Ts&&... args) {
static_assert(std::is_base_of<untyped_actor, Impl>::value ||
(std::is_base_of<blocking_untyped_actor, Impl>::value &&
has_blocking_api_flag(Options)),
......@@ -98,7 +99,8 @@ actor spawn_impl(Ts&&... args) {
using proper_impl = detail::proper_actor<Impl, policies>;
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
CPPA_PUSH_AID(ptr->id());
ptr->launch();
before_launch_fun(ptr.get());
ptr->launch(has_hide_flag(Options));
return ptr;
}
......@@ -121,6 +123,14 @@ struct spawn_fwd<scoped_actor> {
static inline actor fwd(T& arg) { return arg; }
};
template<class Impl, spawn_options Options, typename BeforeLaunch, typename... Ts>
actor spawn_fwd_args(BeforeLaunch before_launch_fun, Ts&&... args) {
return spawn_impl<Impl, Options>(
before_launch_fun,
spawn_fwd<typename util::rm_const_and_ref<Ts>::type>::fwd(
std::forward<Ts>(args))...);
}
/**
* @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments.
......@@ -130,7 +140,9 @@ struct spawn_fwd<scoped_actor> {
*/
template<class Impl, spawn_options Options, typename... Ts>
actor spawn(Ts&&... args) {
return spawn_impl<Impl, Options>(spawn_fwd<typename util::rm_const_and_ref<Ts>::type>::fwd(std::forward<Ts>(args))...);
return spawn_fwd_args<Impl, Options>(
[](local_actor*) { /* no-op as BeforeLaunch callback */ },
std::forward<Ts>(args)...);
}
/**
......@@ -159,19 +171,18 @@ actor spawn(Ts&&... args) {
* @returns An {@link actor} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns.
*/
/*
template<spawn_options Options = no_spawn_options, typename... Ts>
template<spawn_options Options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
auto init_cb = [=](local_actor* ptr) {
ptr->join(grp);
};
return eval_sopts(Options,
get_scheduler()->exec(Options,
init_cb,
std::forward<Ts>(args)...));
using base_class = typename std::conditional<
has_blocking_api_flag(Options),
detail::functor_based_blocking_actor,
detail::functor_based_actor
>::type;
return spawn_fwd_args<base_class, Options>(
[&](local_actor* ptr) { ptr->join(grp); },
std::forward<Ts>(args)...);
}
*/
/**
* @brief Spawns an actor of type @p Impl that immediately joins @p grp.
......@@ -181,14 +192,14 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @returns An {@link actor} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns.
*/
/*
template<class Impl, spawn_options Options, typename... Ts>
actor 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));
return spawn_fwd_args<Impl, Options>(
[&](local_actor* ptr) { ptr->join(grp); },
std::forward<Ts>(args)...);
}
/*
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,
......
......@@ -38,6 +38,8 @@
#include "cppa/mailbox_based.hpp"
#include "cppa/behavior_stack_based.hpp"
#include "cppa/detail/response_future_util.hpp"
namespace cppa {
class untyped_actor;
......@@ -115,7 +117,7 @@ class untyped_actor : public extend<local_actor>::with<mailbox_based,
continue_helper
>::type
then(Fs... fs) {
return then(behavior{(on_arg_match >> std::move(fs))...});
return then(detail::fs2bhvr(m_self, fs...));
}
response_future(const response_future&) = default;
......
......@@ -206,6 +206,8 @@ void abstract_actor::cleanup(std::uint32_t reason) {
for (auto& aptr : mlinks) {
aptr->enqueue({address(), aptr, message_id{}.with_high_priority()}, msg);
}
CPPA_LOGM_DEBUG("cppa::actor", "run " << mattachables.size()
<< "attachables");
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
}
......
......@@ -33,10 +33,21 @@
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/untyped_actor.hpp"
#include "cppa/blocking_untyped_actor.hpp"
namespace cppa {
actor::actor(const std::nullptr_t&) : m_ops(nullptr) { }
actor::actor(actor_proxy* ptr) : m_ops(ptr) { }
actor::actor(untyped_actor* ptr) : m_ops(ptr) { }
actor::actor(blocking_untyped_actor* ptr) : m_ops(ptr) { }
actor::actor(abstract_actor* ptr) : m_ops(ptr) { }
actor::actor(const invalid_actor_t&) : m_ops(nullptr) { }
......
......@@ -64,7 +64,7 @@ intptr_t actor_addr::compare(const actor_addr& other) const {
return compare_impl(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
}
intptr_t actor_addr::compare(const local_actor* other) const {
intptr_t actor_addr::compare(const abstract_actor* other) const {
return compare_impl(m_ops.m_ptr.get(), other);
}
......
......@@ -48,9 +48,11 @@ namespace cppa {
void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
CPPA_REQUIRE(sink != nullptr);
if (!ptr) {
CPPA_LOG_DEBUG("serialize nullptr");
sink->write_value(static_cast<actor_id>(0));
node_id::serialize_invalid(sink);
node_id::host_id_type zero;
std::fill(zero.begin(), zero.end(), 0);
sink->write_value(static_cast<uint32_t>(0)); // actor id
sink->write_value(static_cast<uint32_t>(0)); // process id
sink->write_raw(node_id::host_id_size, zero.data()); // host id
}
else {
// register locally running actors to be able to deserialize them later
......@@ -58,29 +60,31 @@ void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
get_actor_registry()->put(ptr->id(), detail::actor_addr_cast<abstract_actor>(ptr));
}
auto& pinf = ptr->node();
sink->write_value(ptr->id());
sink->write_value(pinf.process_id());
sink->write_raw(node_id::host_id_size, pinf.host_id().data());
sink->write_value(ptr->id()); // actor id
sink->write_value(pinf.process_id()); // process id
sink->write_raw(node_id::host_id_size, pinf.host_id().data()); // host id
}
}
actor_addr actor_namespace::read(deserializer* source) {
CPPA_REQUIRE(source != nullptr);
node_id::host_id_type nid;
auto aid = source->read<uint32_t>();
auto pid = source->read<uint32_t>();
source->read_raw(node_id::host_id_size, nid.data());
// local actor?
auto pinf = node_id::get();
node_id::host_id_type hid;
auto aid = source->read<uint32_t>(); // actor id
auto pid = source->read<uint32_t>(); // process id
source->read_raw(node_id::host_id_size, hid.data()); // host id
auto this_node = node_id::get();
if (aid == 0 && pid == 0) {
// 0:0 identifies an invalid actor
return invalid_actor_addr;
}
else if (pid == pinf->process_id() && nid == pinf->host_id()) {
return actor{get_actor_registry()->get(aid)};
else if (pid == this_node->process_id() && hid == this_node->host_id()) {
// identifies this exact process on this host, ergo: local actor
return get_actor_registry()->get(aid)->address();
}
else {
node_id_ptr tmp = new node_id{pid, nid};
return actor{get_or_put(tmp, aid)};
// identifies a remote actor; create proxy if needed
node_id_ptr tmp = new node_id{pid, hid};
return get_or_put(tmp, aid)->address();
}
}
......
......@@ -45,6 +45,8 @@ using namespace std;
namespace cppa {
actor_proxy::actor_proxy(actor_id mid) : super(mid) { }
actor_proxy::actor_proxy(actor_id mid) : super(mid) {
m_node = node_id::get();
}
} // namespace cppa
......@@ -57,9 +57,9 @@ blocking_untyped_actor::response_future
blocking_untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
auto nri = new_request_id();
get_scheduler()->delayed_send({address(), dest, nri}, rtime, std::move(what));
return {nri.response_id(), this};
return {timed_sync_send_tuple_impl(message_priority::normal, dest, rtime,
std::move(what)),
this};
}
void blocking_untyped_actor::quit(std::uint32_t reason) {
......
......@@ -458,7 +458,7 @@ actor broker::fork_impl(std::function<void (broker*)> fun,
m_io.erase(i);
return {result};
*/
return nullptr;
return invalid_actor;
}
void broker::receive_policy(const connection_handle& hdl,
......
......@@ -140,14 +140,14 @@ class local_broker : public untyped_actor {
behavior make_behavior() override {
return (
on(atom("JOIN"), arg_match) >> [=](const actor_addr& other) {
on(atom("JOIN"), arg_match) >> [=](const actor& other) {
CPPA_LOGC_TRACE("cppa::local_broker", "init$JOIN",
CPPA_TARG(other, to_string));
if (other && m_acquaintances.insert(other).second) {
monitor(other);
}
},
on(atom("LEAVE"), arg_match) >> [=](const actor_addr& other) {
on(atom("LEAVE"), arg_match) >> [=](const actor& other) {
CPPA_LOGC_TRACE("cppa::local_broker", "init$LEAVE",
CPPA_TARG(other, to_string));
if (other && m_acquaintances.erase(other) > 0) {
......@@ -167,7 +167,14 @@ class local_broker : public untyped_actor {
auto sender = last_sender();
CPPA_LOGC_TRACE("cppa::local_broker", "init$DOWN",
CPPA_TARG(sender, to_string));
if (sender) m_acquaintances.erase(sender);
if (sender) {
auto first = m_acquaintances.begin();
auto last = m_acquaintances.end();
auto i = std::find_if(first, last, [=](const actor& a) {
return a == sender;
});
if (i != last) m_acquaintances.erase(i);
}
},
others() >> [=] {
auto msg = last_dequeued();
......@@ -187,13 +194,12 @@ class local_broker : public untyped_actor {
<< " acquaintances; " << CPPA_TSARG(sender)
<< ", " << CPPA_TSARG(what));
for (auto& acquaintance : m_acquaintances) {
auto ptr = detail::actor_addr_cast<abstract_actor>(acquaintance);
ptr->enqueue({sender, ptr}, what);
acquaintance.enqueue({sender, acquaintance}, what);
}
}
local_group_ptr m_group;
set<actor_addr> m_acquaintances;
set<actor> m_acquaintances;
};
......@@ -228,6 +234,7 @@ class local_group_proxy : public local_group {
}
return {who, this};
}
CPPA_LOG_WARNING("channel " << to_string(who) << " already joined");
return {};
}
......
......@@ -140,7 +140,7 @@ void local_actor::forward_message(const actor& dest, message_priority p) {
: m_current_node->mid.with_normal_priority();
detail::raw_access::get(dest)->enqueue({m_current_node->sender, detail::raw_access::get(dest), id}, m_current_node->msg);
// treat this message as asynchronous message from now on
m_current_node->mid = message_id{};
m_current_node->mid = message_id::invalid;
}
void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) {
......@@ -194,4 +194,17 @@ void local_actor::quit(std::uint32_t reason) {
planned_exit_reason(reason);
}
message_id local_actor::timed_sync_send_tuple_impl(message_priority mp,
const actor& dest,
const util::duration& rtime,
any_tuple&& what) {
auto nri = new_request_id();
if (mp == message_priority::high) nri = nri.with_high_priority();
dest.enqueue({address(), dest, nri}, std::move(what));
auto rri = nri.response_id();
get_scheduler()->delayed_send({address(), this, rri}, rtime,
make_any_tuple(atom("TIMEOUT")));
return rri;
}
} // namespace cppa
......@@ -85,7 +85,10 @@ void peer::io_failed(event_bitmask mask) {
auto& children = parent()->get_namespace().proxies(*m_node);
for (auto& kvp : children) {
auto ptr = kvp.second.promote();
send_as(ptr, ptr, atom("KILL_PROXY"), exit_reason::remote_link_unreachable);
if (ptr) {
send_as(ptr, ptr, atom("KILL_PROXY"),
exit_reason::remote_link_unreachable);
}
}
parent()->get_namespace().erase(*m_node);
}
......@@ -236,7 +239,7 @@ void peer::kill_proxy(const actor_addr& sender,
actor_id aid,
std::uint32_t reason) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string)
<< ", " << CPPA_MARG(node, get)
<< ", node = " << (node ? to_string(*node) : "-invalid-")
<< ", " << CPPA_ARG(aid)
<< ", " << CPPA_ARG(reason));
if (!node) {
......
......@@ -49,22 +49,23 @@ inline sync_request_info* new_req_info(actor_addr sptr, message_id id) {
}
sync_request_info::sync_request_info(actor_addr sptr, message_id id)
: next(nullptr), sender(std::move(sptr)), mid(id) { }
: next(nullptr), sender(std::move(sptr)), mid(id) {
}
remote_actor_proxy::remote_actor_proxy(actor_id mid,
const node_id_ptr& pinfo,
node_id_ptr pinfo,
middleman* parent)
: super(mid), m_parent(parent), m_pinf(pinfo) {
: super(mid), m_parent(parent) {
CPPA_REQUIRE(parent != nullptr);
CPPA_LOG_INFO(CPPA_ARG(mid) << ", " << CPPA_TARG(pinfo, to_string)
<< "protocol = " << detail::demangle(typeid(*parent)));
CPPA_LOG_INFO(CPPA_ARG(mid) << ", " << CPPA_TARG(*pinfo, to_string));
m_node = std::move(pinfo);
}
remote_actor_proxy::~remote_actor_proxy() {
auto aid = m_id;
auto node = m_pinf;
auto node = m_node;
auto mm = m_parent;
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(m_pinf)
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(*m_node)
<< ", protocol = " << detail::demangle(typeid(*m_parent)));
mm->run_later([aid, node, mm] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
......@@ -119,7 +120,7 @@ void remote_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg) {
default: break;
}
}
auto node = m_pinf;
auto node = m_node;
auto mm = m_parent;
m_parent->run_later([hdr, msg, node, mm] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
......@@ -142,7 +143,8 @@ void remote_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
m_parent->run_later([_this, reason] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"enqueue$kill_proxy_helper",
"KILL_PROXY with exit reason " << reason);
"KILL_PROXY " << to_string(_this->address())
<< " with exit reason " << reason);
_this->cleanup(reason);
detail::sync_request_bouncer f{reason};
_this->m_pending_requests.close([&](const sync_request_info& e) {
......
......@@ -95,7 +95,8 @@ actor remote_actor(stream_ptr_pair io) {
if (*pinf == *pinfptr) {
// this is a local actor, not a remote actor
CPPA_LOGF_WARNING("remote_actor() called to access a local actor");
return get_actor_registry()->get(remote_aid);
auto ptr = get_actor_registry()->get(remote_aid);
return detail::raw_access::unsafe_cast(ptr.get());
}
auto mm = get_middleman();
struct remote_actor_result { remote_actor_result* next; actor value; };
......
......@@ -185,7 +185,7 @@ void serialize_impl(const actor& ptr, serializer* sink) {
void deserialize_impl(actor& ptr, deserializer* source) {
actor_addr addr;
deserialize_impl(addr, source);
ptr = detail::actor_addr_cast<abstract_actor>(addr);
ptr = detail::raw_access::unsafe_cast(detail::actor_addr_cast<abstract_actor>(addr));
}
void serialize_impl(const group_ptr& ptr, serializer* sink) {
......@@ -223,7 +223,7 @@ void serialize_impl(const channel& ptr, serializer* sink) {
if (aptr != nullptr) {
flag = 1;
sink->write_value(flag);
serialize_impl(actor{aptr}, sink);
serialize_impl(detail::raw_access::unsafe_cast(aptr), sink);
}
else {
auto gptr = group_ptr{dynamic_cast<group*>(rptr)};
......
......@@ -41,13 +41,12 @@ continue_helper& continue_helper::continue_with(behavior::continuation_fun fun)
behavior cpy = *ref_opt;
*ref_opt = cpy.add_continuation(std::move(fun));
}
else CPPA_LOG_ERROR("failed to add continuation");
else { CPPA_LOG_ERROR("failed to add continuation"); }
return *this;
}
void untyped_actor::forward_to(const actor&) {
void untyped_actor::forward_to(const actor& whom) {
forward_message(whom, message_priority::normal);
}
untyped_actor::response_future untyped_actor::sync_send_tuple(const actor& dest,
......@@ -61,9 +60,9 @@ untyped_actor::response_future
untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
auto nri = new_request_id();
get_scheduler()->delayed_send({address(), dest, nri}, rtime, std::move(what));
return {nri.response_id(), this};
return {timed_sync_send_tuple_impl(message_priority::normal, dest, rtime,
std::move(what)),
this};
}
} // namespace cppa
......@@ -29,12 +29,12 @@ void foo() {
}
struct mirror {
mirror(local_actor* self) : m_self(self) { }
mirror(blocking_untyped_actor* self) : m_self(self) { }
template<typename... Ts>
void operator()(Ts&&... args) {
m_self->send(m_self, make_any_tuple(std::forward<Ts>(args)...));
m_self->send(m_self, std::forward<Ts>(args)...);
}
local_actor* m_self;
blocking_untyped_actor* m_self;
};
int main() {
......
......@@ -11,6 +11,8 @@
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/raw_access.hpp"
using namespace std;
using namespace cppa;
......@@ -33,8 +35,8 @@ void reflector(untyped_actor* self) {
void spawn5_server_impl(untyped_actor* self, actor client, group_ptr grp) {
CPPA_LOGF_TRACE(CPPA_TARG(client, to_string)
<< ", " << CPPA_TARG(grp, to_string));
//FIXME spawn_in_group(grp, reflector);
//FIXME spawn_in_group(grp, reflector);
self->spawn_in_group(grp, reflector);
self->spawn_in_group(grp, reflector);
CPPA_LOGF_INFO("send {'Spawn5'} and await {'ok', actor_vector}");
self->sync_send(client, atom("Spawn5"), grp).then(
on(atom("ok"), arg_match) >> [=](const actor_vector& vec) {
......@@ -80,6 +82,8 @@ void spawn5_server_impl(untyped_actor* self, actor client, group_ptr grp) {
},
after(chrono::seconds(2)) >> [=] {
CPPA_UNEXPECTED_TOUT();
CPPA_LOGF_ERROR("did only receive " << *downs
<< " down messages");
self->quit(exit_reason::unhandled_exception);
}
);
......@@ -87,6 +91,8 @@ void spawn5_server_impl(untyped_actor* self, actor client, group_ptr grp) {
},
after(std::chrono::seconds(2)) >> [=] {
CPPA_UNEXPECTED_TOUT();
CPPA_LOGF_ERROR("did only receive " << *replies
<< " responses to 'Hello reflectors!'");
self->quit(exit_reason::unhandled_exception);
}
);
......@@ -112,11 +118,11 @@ void spawn5_client(untyped_actor* self) {
CPPA_LOGF_INFO("received {'GetGroup'}");
return group::get("local", "foobar");
},
on(atom("Spawn5"), arg_match) >> [=](const group_ptr&) -> any_tuple {
on(atom("Spawn5"), arg_match) >> [=](const group_ptr& grp) -> any_tuple {
CPPA_LOGF_INFO("received {'Spawn5'}");
actor_vector vec;
for (int i = 0; i < 5; ++i) {
//FIXME vec.push_back(spawn_in_group(grp, reflector));
vec.push_back(spawn_in_group(grp, reflector));
}
return make_any_tuple(atom("ok"), std::move(vec));
},
......@@ -130,9 +136,8 @@ void spawn5_client(untyped_actor* self) {
} // namespace <anonymous>
template<typename F>
void await_down(actor, F) {
/*
become (
void await_down(untyped_actor* self, actor ptr, F continuation) {
self->become (
on(atom("DOWN"), arg_match) >> [=](uint32_t) -> bool {
if (self->last_sender() == ptr) {
continuation();
......@@ -141,7 +146,6 @@ void await_down(actor, F) {
return false; // not the 'DOWN' message we are waiting for
}
);
*/
}
static constexpr size_t num_pings = 10;
......@@ -164,7 +168,7 @@ class client : public untyped_actor {
return (
on(atom("PingPtr"), arg_match) >> [=](const actor& ping) {
auto pptr = spawn<monitored+detached+blocking_api>(pong, ping);
await_down(pptr, [=] {
await_down(this, pptr, [=] {
send_sync_msg();
});
}
......@@ -199,7 +203,7 @@ class client : public untyped_actor {
sync_send(m_server, atom("GClient")).then(
on(atom("GClient"), arg_match) >> [=](actor gclient) {
auto s5a = spawn<monitored>(spawn5_server, gclient, false);
await_down(s5a, [=]{
await_down(this, s5a, [=]{
test_group_comm_inverted();
});
}
......@@ -213,7 +217,7 @@ class client : public untyped_actor {
auto cptr = last_sender();
auto s5c = spawn<monitored>(spawn5_client);
// set next behavior
await_down(s5c, [=] {
await_down(this, s5c, [=] {
CPPA_CHECKPOINT();
quit();
});
......@@ -246,7 +250,7 @@ class server : public untyped_actor {
CPPA_LOGF_INFO("spawn event-based ping actor");
auto pptr = spawn<monitored>(event_based_ping, num_pings);
CPPA_LOGF_INFO("wait until spawned ping actor is done");
await_down(pptr, [=] {
await_down(this, pptr, [=] {
CPPA_CHECK_EQUAL(pongs(), num_pings);
await_sync_msg();
});
......@@ -286,10 +290,12 @@ class server : public untyped_actor {
CPPA_PRINT("test group communication via network");
become (
on(atom("GClient")) >> [=]() -> any_tuple {
CPPA_CHECKPOINT();
auto cptr = last_sender();
auto s5c = spawn<monitored>(spawn5_client);
await_down(s5c, [=] {
//test_group_comm_inverted(cptr);
await_down(this, s5c, [=] {
CPPA_CHECKPOINT();
test_group_comm_inverted(detail::raw_access::unsafe_cast(cptr));
});
return make_any_tuple(atom("GClient"), s5c);
}
......@@ -300,7 +306,7 @@ class server : public untyped_actor {
CPPA_PRINT("test group communication via network (inverted setup)");
sync_send(cptr, atom("GClient")).then (
on(atom("GClient"), arg_match) >> [=](actor gclient) {
await_down(spawn<monitored>(spawn5_server, gclient, true), [=] {
await_down(this, spawn<monitored>(spawn5_server, gclient, true), [=] {
CPPA_CHECKPOINT();
quit();
});
......
......@@ -77,6 +77,7 @@ struct B : popular_actor {
struct C : sb_actor<C> {
behavior init_state = (
on(atom("gogo")) >> [=]() -> atom_value {
CPPA_CHECKPOINT();
quit();
return atom("gogogo");
}
......@@ -206,10 +207,13 @@ void test_sync_send() {
auto mirror = spawn<sync_mirror>();
bool continuation_called = false;
self->sync_send(mirror, 42)
.await([](int value) { CPPA_CHECK_EQUAL(value, 42); });
.await([&](int value) {
continuation_called = true;
CPPA_CHECK_EQUAL(value, 42);
});
CPPA_CHECK_EQUAL(continuation_called, true);
self->send_exit(mirror, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
auto await_success_message = [&] {
self->receive (
......@@ -224,11 +228,11 @@ void test_sync_send() {
self->send(self->spawn<A, monitored>(self), atom("go"), spawn<B>(spawn<C>()));
await_success_message();
CPPA_CHECKPOINT();
await_all_actors_done();
self->await_all_other_actors_done();
self->send(self->spawn<A, monitored>(self), atom("go"), spawn<D>(spawn<C>()));
await_success_message();
CPPA_CHECKPOINT();
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
self->timed_sync_send(self, std::chrono::milliseconds(50), atom("NoWay")).await(
on(atom("TIMEOUT")) >> CPPA_CHECKPOINT_CB(),
......@@ -265,7 +269,7 @@ void test_sync_send() {
self->on_sync_failure(CPPA_UNEXPECTED_MSG_CB());
self->sync_send(c, atom("gogo")).await(CPPA_CHECKPOINT_CB());
self->send_exit(c, exit_reason::user_shutdown);
await_all_actors_done();
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
// test use case 3
......@@ -275,13 +279,16 @@ void test_sync_send() {
self->become(on(atom("request")) >> []{ return atom("response"); });
});
// first 'idle', then 'request'
send_as(w, s, atom("idle"));
anon_send(s, atom("idle"), w);
self->sync_send(s, atom("request")).await(
on(atom("response")) >> [=] {
CPPA_CHECKPOINT();
CPPA_CHECK_EQUAL(self->last_sender(), w);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
others() >> [&] {
CPPA_PRINTERR("unexpected message: "
<< to_string(self->last_dequeued()));
}
);
// first 'request', then 'idle'
auto handle = self->sync_send(s, atom("request"));
......
......@@ -84,6 +84,7 @@ int main() {
"@ac_hdl", // io::accept_handle
"@cn_hdl", // io::connection_handle
"@atom", // atom_value
"@addr", // actor address
"@tuple", // any_tuple
"@header", // message_header
"@actor", // actor_ptr
......@@ -117,13 +118,25 @@ int main() {
CPPA_CHECK(expected_equals_found);
}
if (!expected_equals_found) {
CPPA_PRINT("found:");
for (const std::string& tname : found) {
CPPA_PRINT(" - " << tname);
}
CPPA_PRINT("expected: ");
for (const std::string& tname : expected) {
CPPA_PRINT(" - " << tname);
std::string(41, ' ');
std::ostringstream oss(std::string(41, ' '));
oss.seekp(0);
oss << "found (" << found.size() << ")";
oss.seekp(22);
oss << "expected (" << found.size() << ")";
std::string lhs;
std::string rhs;
CPPA_PRINT(oss.str());
CPPA_PRINT(std::string(41, '-'));
auto fi = found.begin();
auto fe = found.end();
auto ei = expected.begin();
auto ee = expected.end();
while (fi != fe || ei != ee) {
if (fi != fe) lhs = *fi++; else lhs.clear();
if (ei != ee) rhs = *ei++; else rhs.clear();
lhs.resize(20, ' ');
CPPA_PRINT(lhs << "| " << rhs);
}
}
......
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