Commit 70c87f40 authored by Dominik Charousset's avatar Dominik Charousset

re-implemted broker, fixed broker unit test

parent 8a3e7944
......@@ -48,7 +48,13 @@ class actor_proxy;
class untyped_actor;
class blocking_untyped_actor;
namespace detail { class raw_access; }
namespace io {
class broker;
} // namespace io
namespace detail {
class raw_access;
} // namespace detail
struct invalid_actor_t { constexpr invalid_actor_t() { } };
......@@ -68,7 +74,8 @@ class actor : util::comparable<actor> {
template<typename T>
actor(intrusive_ptr<T> ptr,
typename std::enable_if<
std::is_base_of<actor_proxy, T>::value
std::is_base_of<io::broker, T>::value
|| 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)
......
......@@ -31,6 +31,8 @@
#ifndef CPPA_BEHAVIOR_STACK_BASED_HPP
#define CPPA_BEHAVIOR_STACK_BASED_HPP
#include "cppa/message_id.hpp"
#include "cppa/detail/behavior_stack.hpp"
namespace cppa {
......@@ -74,10 +76,17 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
template<class Base, class Subtype>
class behavior_stack_based : public Base {
typedef Base super;
public:
typedef behavior_stack_based combined_type;
template <typename... Ts>
behavior_stack_based(Ts&&... args)
: super(std::forward<Ts>(args)...), m_has_timeout(false)
, m_timeout_id(0) { }
inline void unbecome() {
m_bhvr_stack.pop_async_back();
}
......@@ -102,10 +111,6 @@ class behavior_stack_based : public Base {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
virtual void become_waiting_for(behavior bhvr, message_id mf) = 0;
virtual void do_become(behavior bhvr, bool discard_old) = 0;
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
......@@ -115,10 +120,6 @@ class behavior_stack_based : public Base {
return m_bhvr_stack.back();
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
......@@ -131,11 +132,75 @@ class behavior_stack_based : public Base {
m_bhvr_stack.erase(mid);
}
void become_waiting_for(behavior bhvr, message_id mf) {
//CPPA_LOG_TRACE(CPPA_MARG(mf, integer_value));
if (bhvr.timeout().valid()) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
void do_become(behavior bhvr, bool discard_old) {
//CPPA_LOG_TRACE(CPPA_ARG(discard_old));
//if (discard_old) m_bhvr_stack.pop_async_back();
//m_bhvr_stack.push_back(std::move(bhvr));
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->reset_timeout();
if (bhvr.timeout().valid()) {
//CPPA_LOG_DEBUG("request timeout: " << bhvr.timeout().to_string());
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr));
}
// timeout handling
void request_timeout(const util::duration& d) {
if (d.valid()) {
m_has_timeout = true;
auto tid = ++m_timeout_id;
auto msg = make_any_tuple(atom("SYNC_TOUT"), tid);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue({this->address(), this}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else this->delayed_send_tuple(this, d, std::move(msg));
}
else m_has_timeout = false;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) const {
return m_has_timeout && m_timeout_id == timeout_id;
}
inline bool is_active_timeout(std::uint32_t tid) const {
return waits_for_timeout(tid);
}
inline void reset_timeout() {
m_has_timeout = false;
}
inline void handle_timeout(behavior& bhvr, std::uint32_t timeout_id) {
CPPA_REQUIRE(m_timeout_id == timeout_id);
m_has_timeout = false;
bhvr.handle_timeout();
}
protected:
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
bool m_has_timeout;
std::uint32_t m_timeout_id;
};
} // namespace cppa
......
......@@ -42,6 +42,8 @@
namespace cppa {
class actor;
struct invalid_actor_t;
namespace detail { class raw_access; }
/**
......@@ -64,6 +66,8 @@ class channel : util::comparable<channel>
channel(const std::nullptr_t&);
channel(const invalid_actor_t&);
template<typename T>
channel(intrusive_ptr<T> ptr, typename std::enable_if<std::is_base_of<abstract_channel, T>::value>::type* = 0) : m_ptr(ptr) { }
......
......@@ -525,17 +525,9 @@ actor remote_actor(io::stream_ptr_pair connection);
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
using namespace policy;
using ps = policies<middleman_scheduling, not_prioritizing,
no_resume, cooperative_scheduling>;
using proper_impl = detail::proper_actor<Impl, ps>;
auto ptr = make_counted<proper_impl>(std::move(in), std::move(out),
std::forward<Ts>(args)...);
ptr->launch();
return ptr;
actor spawn_io(Ts&&... args) {
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
return {io::init_and_launch(std::move(ptr))};
}
/**
......@@ -545,12 +537,15 @@ actor spawn_io(io::input_stream_ptr in,
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>>
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out) {
return spawn_io<io::default_broker>(std::move(fun), std::move(in),
std::move(out));
io::output_stream_ptr out,
Ts&&... args) {
auto ptr = io::broker::from(std::move(fun), std::move(in), std::move(out),
std::forward<Ts>(args)...);
return {io::init_and_launch(std::move(ptr))};
}
/*
......@@ -562,17 +557,26 @@ actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) {
*/
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>>
actor spawn_io(F fun, const std::string& host, uint16_t port) {
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host.c_str(), port);
return spawn_io(std::move(fun), ptr, ptr);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
}
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>>
actor spawn_io_server(F fun, uint16_t port) {
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
static_assert(!has_detach_flag(Options),
"brokers cannot be detached");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag");
using namespace std;
return spawn_io(move(fun), io::ipv4_acceptor::create(port));
auto ptr = io::broker::from(move(fun),
io::ipv4_acceptor::create(port),
forward<Ts>(args)...);
return {io::init_and_launch(move(ptr))};
}
/**
......
......@@ -39,10 +39,6 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D
return this->m_mailbox;
}
mailbox_element* dummy_node() {
return &this->m_dummy_node;
}
// member functions from scheduling policy
typedef typename Policies::scheduling_policy::timeout_type timeout_type;
......@@ -167,10 +163,15 @@ class proper_actor : public proper_actor_base<Base,
public:
static_assert(std::is_base_of<local_actor, Base>::value,
"Base is not derived from local_actor");
// this is the nonblocking version of proper_actor; it assumes
// that Base is derived from local_actor and uses the
// behavior_stack_based mixin
template <typename... Ts>
proper_actor(Ts&&... args)
: super(std::forward<Ts>(args)...), m_has_timeout(false)
, m_timeout_id(0) { }
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
inline void launch(bool is_hidden) {
CPPA_LOG_TRACE("");
......@@ -200,82 +201,17 @@ class proper_actor : public proper_actor_base<Base,
CPPA_LOG_DEBUG(std::distance(this->cache_begin(), e)
<< " elements in cache");
for (auto i = this->cache_begin(); i != e; ++i) {
if (this->invoke_policy().invoke_message(this, *i, bhvr, mid)) {
auto res = this->invoke_policy().invoke_message(this, *i, bhvr, mid);
if (res || !*i) {
this->cache_erase(i);
return true;
if (res) return true;
// start anew, because we have invalidated our iterators now
return invoke_message_from_cache();
}
}
return false;
}
// implement pure virtual functions from behavior_stack_based
void become_waiting_for(behavior bhvr, message_id mf) override {
CPPA_LOG_TRACE(CPPA_MARG(mf, integer_value));
if (bhvr.timeout().valid()) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
void do_become(behavior bhvr, bool discard_old) override {
CPPA_LOG_TRACE(CPPA_ARG(discard_old));
//if (discard_old) m_bhvr_stack.pop_async_back();
//m_bhvr_stack.push_back(std::move(bhvr));
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->reset_timeout();
if (bhvr.timeout().valid()) {
CPPA_LOG_DEBUG("request timeout: " << bhvr.timeout().to_string());
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr));
}
// timeout handling
void request_timeout(const util::duration& d) {
if (d.valid()) {
m_has_timeout = true;
auto tid = ++m_timeout_id;
auto msg = make_any_tuple(atom("SYNC_TOUT"), tid);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue({this->address(), this}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else this->delayed_send_tuple(this, d, std::move(msg));
}
else m_has_timeout = false;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) const {
return m_has_timeout && m_timeout_id == timeout_id;
}
inline bool is_active_timeout(std::uint32_t tid) const {
return waits_for_timeout(tid);
}
inline void reset_timeout() {
m_has_timeout = false;
}
inline void handle_timeout(behavior& bhvr, std::uint32_t timeout_id) {
CPPA_REQUIRE(m_timeout_id == timeout_id);
m_has_timeout = false;
bhvr.handle_timeout();
}
private:
bool m_has_timeout;
std::uint32_t m_timeout_id;
};
// for blocking actors, there's one more member function to implement
......
......@@ -45,6 +45,7 @@
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp"
#include "cppa/policy/not_prioritizing.hpp"
#include "cppa/policy/sequential_invoke.hpp"
namespace cppa {
......@@ -54,7 +55,7 @@ class broker;
typedef intrusive_ptr<broker> broker_ptr;
local_actor_ptr init_and_launch(broker_ptr);
broker_ptr init_and_launch(broker_ptr);
/**
* @brief A broker mediates between a libcppa-based actor system
......@@ -78,20 +79,20 @@ class broker : public extend<local_actor>::with<behavior_stack_based> {
friend class doorman;
friend class continuation;
friend local_actor_ptr init_and_launch(broker_ptr);
friend broker_ptr init_and_launch(broker_ptr);
broker() = delete;
public:
~broker();
enum policy_flag { at_least, at_most, exactly };
void enqueue(const message_header& hdr, any_tuple msg);
bool initialized() const;
void quit(std::uint32_t reason = exit_reason::normal) override;
void receive_policy(const connection_handle& hdl,
broker::policy_flag policy,
size_t buffer_size);
......@@ -102,7 +103,6 @@ class broker : public extend<local_actor>::with<behavior_stack_based> {
void write(const connection_handle& hdl, util::buffer&& buf);
/*
template<typename F, typename... Ts>
static broker_ptr from(F fun,
input_stream_ptr in,
......@@ -127,7 +127,6 @@ class broker : public extend<local_actor>::with<behavior_stack_based> {
std::forward<Ts>(args)...),
std::move(in));
}
*/
template<typename F, typename... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... args) {
......@@ -168,12 +167,14 @@ class broker : public extend<local_actor>::with<behavior_stack_based> {
actor fork_impl(std::function<void (broker*)> fun,
connection_handle hdl);
//static broker_ptr from_impl(std::function<void (broker*)> fun,
// input_stream_ptr in,
// output_stream_ptr out);
static broker_ptr from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
void invoke_message(const message_header& hdr, any_tuple msg);
bool invoke_message_from_cache();
void erase_io(int id);
void erase_acceptor(int id);
......@@ -187,7 +188,8 @@ class broker : public extend<local_actor>::with<behavior_stack_based> {
std::map<accept_handle, doorman_pointer> m_accept;
std::map<connection_handle, scribe_pointer> m_io;
policy::sequential_invoke m_invoke;
policy::not_prioritizing m_priority_policy;
policy::sequential_invoke m_invoke_policy;
};
......@@ -197,7 +199,11 @@ class default_broker : public broker {
typedef std::function<void (broker*)> function_type;
default_broker(input_stream_ptr in, output_stream_ptr out, function_type f);
default_broker(function_type f, input_stream_ptr in, output_stream_ptr out);
default_broker(function_type f, scribe_pointer ptr);
default_broker(function_type f, acceptor_uptr ptr);
behavior make_behavior() override;
......
......@@ -391,6 +391,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void cleanup(std::uint32_t reason);
mailbox_element* dummy_node() {
return &m_dummy_node;
}
protected:
template<typename... Ts>
......
......@@ -61,8 +61,6 @@ actor spawn_impl(BeforeLaunch before_launch_fun, Ts&&... args) {
"blocking_api_flag is missing");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag");
CPPA_LOGF_TRACE("spawn " << detail::demangle<Impl>());
// runtime check wheter context_switching_resume can be used,
// i.e., add the detached flag if libcppa compiled without fiber support
......
......@@ -32,7 +32,7 @@
#include "cppa/config.hpp"
//#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/singletons.hpp"
#include "cppa/util/scope_guard.hpp"
......@@ -58,13 +58,22 @@ constexpr size_t default_max_buffer_size = 65535;
} // namespace <anonymous>
default_broker::default_broker(input_stream_ptr in,
output_stream_ptr out,
function_type f)
default_broker::default_broker(function_type f,
input_stream_ptr in,
output_stream_ptr out)
: broker(std::move(in), std::move(out)), m_fun(std::move(f)) { }
default_broker::default_broker(function_type f, scribe_pointer ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
default_broker::default_broker(function_type f, acceptor_uptr ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
behavior default_broker::make_behavior() {
enqueue({invalid_actor_addr, channel{this}}, make_any_tuple(atom("INITMSG")));
CPPA_PUSH_AID(id());
CPPA_LOG_TRACE("");
enqueue({invalid_actor_addr, channel{this}},
make_any_tuple(atom("INITMSG")));
return (
on(atom("INITMSG")) >> [=] {
unbecome();
......@@ -81,6 +90,8 @@ class broker::continuation {
: m_self(move(ptr)), m_hdr(hdr), m_data(move(msg)) { }
inline void operator()() {
CPPA_PUSH_AID(m_self->id());
CPPA_LOG_TRACE("");
m_self->invoke_message(m_hdr, move(m_data));
}
......@@ -126,7 +137,8 @@ class broker::servant : public continuable {
if (!m_disconnected) {
m_disconnected = true;
if (m_broker->exit_reason() == exit_reason::not_exited) {
//TODO: m_broker->invoke_message(nullptr, disconnect_message());
m_broker->invoke_message({invalid_actor_addr, invalid_actor},
disconnect_message());
}
}
}
......@@ -275,7 +287,11 @@ class broker::doorman : public broker::servant {
};
void broker::invoke_message(const message_header& hdr, any_tuple msg) {
if (exit_reason() != exit_reason::not_exited || m_bhvr_stack.empty()) {
CPPA_LOG_TRACE(CPPA_TARG(msg, to_string));
if (planned_exit_reason() != exit_reason::not_exited || bhvr_stack().empty()) {
CPPA_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = " << planned_exit_reason()
<< ", bhvr_stack().empty() = " << bhvr_stack().empty());
if (hdr.id.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(hdr.sender, hdr.id);
......@@ -287,30 +303,30 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
m_dummy_node.msg = move(msg);
m_dummy_node.mid = hdr.id;
try {
/*TODO:
auto bhvr = m_bhvr_stack.back();
switch (m_invoke.handle_message(this,
&m_dummy_node,
bhvr,
m_bhvr_stack.back_id())) {
auto bhvr = bhvr_stack().back();
auto mid = bhvr_stack().back_id();
switch (m_invoke_policy.handle_message(this, &m_dummy_node, bhvr, mid)) {
case policy::hm_msg_handled: {
if ( not m_bhvr_stack.empty()
&& bhvr.as_behavior_impl() == m_bhvr_stack.back().as_behavior_impl()) {
//TODO: m_invoke.invoke_from_cache(this, bhvr, m_bhvr_stack.back_id());
CPPA_LOG_DEBUG("handle_message returned hm_msg_handled");
while ( !bhvr_stack().empty()
&& planned_exit_reason() == exit_reason::not_exited
&& invoke_message_from_cache()) {
// rinse and repeat
}
break;
}
case policy::hm_drop_msg:
CPPA_LOG_DEBUG("handle_message returned hm_drop_msg");
break;
case policy::hm_skip_msg:
case policy::hm_cache_msg: {
CPPA_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
auto e = mailbox_element::create(hdr, move(m_dummy_node.msg));
//TODO: m_invoke.add_to_cache(e);
m_priority_policy.push_to_cache(unique_mailbox_element_pointer{e});
break;
}
default: CPPA_CRITICAL("illegal result of handle_message");
}
*/
}
catch (std::exception& e) {
CPPA_LOG_ERROR("broker killed due to an unhandled exception: "
......@@ -326,6 +342,33 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
// restore dummy node
m_dummy_node.sender = actor_addr{};
m_dummy_node.msg.reset();
// cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason());
}
else if (bhvr_stack().empty()) {
CPPA_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason");
quit(exit_reason::normal);
cleanup(planned_exit_reason());
}
}
bool broker::invoke_message_from_cache() {
CPPA_LOG_TRACE("");
auto bhvr = bhvr_stack().back();
auto mid = bhvr_stack().back_id();
auto e = m_priority_policy.cache_end();
CPPA_LOG_DEBUG(std::distance(m_priority_policy.cache_begin(), e)
<< " elements in cache");
for (auto i = m_priority_policy.cache_begin(); i != e; ++i) {
auto res = m_invoke_policy.invoke_message(this, *i, bhvr, mid);
if (res || !*i) {
m_priority_policy.cache_erase(i);
if (res) return true;
return invoke_message_from_cache();
}
}
return false;
}
void broker::enqueue(const message_header& hdr, any_tuple msg) {
......@@ -336,12 +379,6 @@ bool broker::initialized() const {
return true;
}
void broker::quit(std::uint32_t reason) {
//cleanup(reason);
m_bhvr_stack.clear();
super::quit(reason);
}
void broker::init_broker() {
// acquire implicit reference count held by the middleman
ref();
......@@ -387,12 +424,11 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) {
buf.clear();
}
local_actor_ptr init_and_launch(broker_ptr ptr) {
//ptr->init();
broker_ptr init_and_launch(broker_ptr ptr) {
CPPA_PUSH_AID(ptr->id());
CPPA_LOGF_TRACE("init and launch actor with id " << ptr->id());
// continue reader only if not inherited from default_broker_impl
CPPA_LOGF_WARNING_IF(!ptr->has_behavior(), "broker w/o behavior spawned");
auto mm = get_middleman();
if (ptr->has_behavior()) {
mm->run_later([=] {
CPPA_LOGC_TRACE("NONE", "init_and_launch::run_later_functor", "");
CPPA_LOGF_WARNING_IF(ptr->m_io.empty() && ptr->m_accept.empty(),
......@@ -407,11 +443,13 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
for (auto& kvp : ptr->m_accept)
mm->continue_reader(kvp.second.get());
});
}
// exec initialization code
auto bhvr = ptr->make_behavior();
if (bhvr) ptr->become(std::move(bhvr));
CPPA_LOGF_WARNING_IF(!ptr->has_behavior(), "broker w/o behavior spawned");
return ptr;
}
/*
broker_ptr broker::from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out) {
......@@ -421,8 +459,6 @@ broker_ptr broker::from_impl(std::function<void (broker*)> fun,
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
return make_counted<default_broker>(move(fun), move(in));
}
*/
void broker::erase_io(int id) {
m_io.erase(connection_handle::from_int(id));
......@@ -448,17 +484,18 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) {
actor broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) {
CPPA_LOG_TRACE(CPPA_MARG(hdl, id));
auto i = m_io.find(hdl);
if (i == m_io.end()) throw std::invalid_argument("invalid handle");
/*FIXME
if (i == m_io.end()) {
CPPA_LOG_ERROR("invalid handle");
throw std::invalid_argument("invalid handle");
}
scribe* sptr = i->second.get(); // non-owning pointer
auto result = make_counted<default_broker>(move(fun), move(i->second));
init_and_launch(result);
sptr->set_broker(result); // set new broker
m_io.erase(i);
return {result};
*/
return invalid_actor;
}
void broker::receive_policy(const connection_handle& hdl,
......@@ -468,5 +505,9 @@ void broker::receive_policy(const connection_handle& hdl,
if (i != m_io.end()) i->second->receive_policy(policy, buffer_size);
}
broker::~broker() {
CPPA_LOG_TRACE("");
}
} // namespace io
} // namespace cppa
......@@ -38,6 +38,8 @@ namespace cppa {
channel::channel(const std::nullptr_t&) : m_ptr(nullptr) { }
channel::channel(const invalid_actor_t&) : m_ptr(nullptr) { }
channel::channel(const actor& other) : m_ptr(detail::raw_access::get(other)) { }
intptr_t channel::compare(const abstract_channel* lhs, const abstract_channel* rhs) {
......
......@@ -40,9 +40,11 @@ using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void ping(cppa::untyped_actor* self, size_t num_pings) {
CPPA_CHECKPOINT();
auto count = std::make_shared<size_t>(0);
self->become (
on(atom("kickoff"), arg_match) >> [=](const actor& pong) {
CPPA_CHECKPOINT();
self->send(pong, atom("ping"), 1);
self->become (
on(atom("pong"), arg_match)
......@@ -58,9 +60,11 @@ void ping(cppa::untyped_actor* self, size_t num_pings) {
}
void pong(cppa::untyped_actor* self) {
CPPA_CHECKPOINT();
self->become (
on(atom("ping"), arg_match)
>> [=](int value) -> cow_tuple<atom_value, int> {
CPPA_CHECKPOINT();
self->monitor(self->last_sender());
// set next behavior
self->become (
......@@ -80,17 +84,20 @@ void pong(cppa::untyped_actor* self) {
}
void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
CPPA_CHECKPOINT();
self->monitor(buddy);
if (self->num_connections() == 0) {
cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1");
}
auto write = [=](atom_value type, int value) {
CPPA_LOGF_DEBUG("write: " << value);
self->write(hdl, sizeof(type), &type);
self->write(hdl, sizeof(value), &value);
};
self->become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
CPPA_LOGF_INFO("received IO_closed");
self->quit();
},
on(atom("IO_read"), arg_match) >> [=](io::connection_handle, const util::buffer& buf) {
......@@ -114,6 +121,7 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
}
void peer_acceptor(io::broker* self, const actor& buddy) {
CPPA_CHECKPOINT();
self->become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) {
CPPA_CHECKPOINT();
......@@ -135,8 +143,7 @@ int main(int argc, char** argv) {
CPPA_CHECKPOINT();
auto p = spawn(ping, 10);
CPPA_CHECKPOINT();
//FIXME auto cl = spawn_io(peer, "localhost", port, p);
actor cl;
auto cl = spawn_io(peer, "localhost", port, p);
CPPA_CHECKPOINT();
send_as(nullptr, p, atom("kickoff"), cl);
CPPA_CHECKPOINT();
......@@ -155,7 +162,7 @@ int main(int argc, char** argv) {
uint16_t port = 4242;
for (;;) {
try {
//FIXME: spawn_io_server(peer_acceptor, port, p);
spawn_io_server(peer_acceptor, port, p);
CPPA_CHECKPOINT();
ostringstream oss;
oss << app_path << " mode=client port=" << port << " &>/dev/null";
......@@ -173,11 +180,11 @@ int main(int argc, char** argv) {
await_all_actors_done();
CPPA_CHECKPOINT();
shutdown();
return CPPA_TEST_RESULT();
}
catch (bind_failure&) {
// try next port
++port;
}
}
return CPPA_TEST_RESULT();
}
......@@ -337,7 +337,7 @@ struct slave : untyped_actor {
};
void test_serial_reply() {
auto mirror_behavior = [=](untyped_actor* self, int num) {
auto mirror_behavior = [=](untyped_actor* self) {
self->become(others() >> [=]() -> any_tuple {
CPPA_LOGF_INFO("return self->last_dequeued()");
return self->last_dequeued();
......@@ -346,11 +346,11 @@ void test_serial_reply() {
auto master = spawn([=](untyped_actor* self) {
cout << "ID of master: " << self->id() << endl;
// spawn 5 mirror actors
auto c0 = self->spawn<linked>(mirror_behavior, 0);
auto c1 = self->spawn<linked>(mirror_behavior, 1);
auto c2 = self->spawn<linked>(mirror_behavior, 2);
auto c3 = self->spawn<linked>(mirror_behavior, 3);
auto c4 = self->spawn<linked>(mirror_behavior, 4);
auto c0 = self->spawn<linked>(mirror_behavior);
auto c1 = self->spawn<linked>(mirror_behavior);
auto c2 = self->spawn<linked>(mirror_behavior);
auto c3 = self->spawn<linked>(mirror_behavior);
auto c4 = self->spawn<linked>(mirror_behavior);
self->become (
on(atom("hi there")) >> [=]() -> continue_helper {
CPPA_LOGF_INFO("received 'hi there'");
......
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