Commit 3b8cea71 authored by Dominik Charousset's avatar Dominik Charousset

Further design changes for upcoming 0.9 release

parent 6773bedb
......@@ -139,6 +139,7 @@ set(LIBCPPA_SRC
src/broker.cpp
src/buffer.cpp
src/channel.cpp
src/common_actor_ops.cpp
src/context_switching_actor.cpp
src/continuable.cpp
src/decorated_tuple.cpp
......@@ -188,6 +189,7 @@ set(LIBCPPA_SRC
src/shared_spinlock.cpp
src/singleton_manager.cpp
src/string_serialization.cpp
src/sync_request_bouncer.cpp
src/thread_mapped_actor.cpp
src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp
......
......@@ -84,20 +84,6 @@ class abstract_actor : public abstract_channel {
*/
bool attach(attachable_ptr ptr);
/**
* @brief Convenience function that attaches the functor
* or function @p f to this actor.
*
* The actor executes <tt>f()</tt> on exit, or immediatley
* if it already finished execution.
* @param f A functor, function or lambda expression that's executed
* if the actor finishes execution.
* @returns @c true if @p f was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*/
template<typename F>
bool attach_functor(F&& f);
/**
* @brief Returns the address of this actor.
*/
......@@ -236,27 +222,6 @@ inline bool abstract_actor::exited() const {
return exit_reason() != exit_reason::not_exited;
}
template<class F>
struct functor_attachable : attachable {
F m_functor;
template<typename T>
inline functor_attachable(T&& arg) : m_functor(std::forward<T>(arg)) { }
void actor_exited(std::uint32_t reason) { m_functor(reason); }
bool matches(const attachable::token&) { return false; }
};
template<typename F>
bool abstract_actor::attach_functor(F&& f) {
typedef typename util::rm_const_and_ref<F>::type f_type;
typedef functor_attachable<f_type> impl;
return attach(attachable_ptr{new impl(std::forward<F>(f))});
}
inline const node_id& abstract_actor::node() const {
return *m_node;
}
......
......@@ -36,55 +36,60 @@
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/common_actor_ops.hpp"
#include "cppa/util/comparable.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
class channel;
class any_tuple;
class actor_addr;
class local_actor;
class message_header;
namespace detail { class raw_access; }
/**
* @brief Identifies an untyped actor.
*/
class actor : util::comparable<actor> {
friend class channel;
friend class actor_addr;
friend class local_actor;
friend class detail::raw_access;
public:
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_ptr(ptr) { }
actor(intrusive_ptr<T> ptr, typename std::enable_if<std::is_base_of<abstract_actor, T>::value>::type* = 0) : m_ops(ptr) { }
actor(abstract_actor*);
actor_id id() const;
explicit inline operator bool() const;
actor_addr address() const;
explicit operator bool() const;
bool operator!() const;
inline bool operator!() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
bool is_remote() const;
inline common_actor_ops* operator->() const {
// this const cast is safe, because common_actor_ops cannot be
// modified anyways and the offered operations are intended to
// be called on const elements
return const_cast<common_actor_ops*>(&m_ops);
}
intptr_t compare(const actor& other) const;
private:
intrusive_ptr<abstract_actor> m_ptr;
common_actor_ops m_ops;
};
inline actor::operator bool() const {
return static_cast<bool>(m_ops.m_ptr);
}
inline bool actor::operator!() const {
return !static_cast<bool>(*this);
}
} // namespace cppa
#endif // CPPA_ACTOR_HPP
......@@ -37,27 +37,26 @@
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/common_actor_ops.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
class actor;
class node_id;
class actor_addr;
class local_actor;
namespace detail { class raw_access; }
namespace detail { template<typename T> T* actor_addr_cast(const actor_addr&); }
struct invalid_actor_addr_t { constexpr invalid_actor_addr_t() { } };
constexpr struct invalid_actor_addr_t { constexpr invalid_actor_addr_t() { } } invalid_actor_addr;
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*> {
friend class abstract_actor;
template<typename T>
friend T* detail::actor_addr_cast(const actor_addr&);
friend class detail::raw_access;
public:
......@@ -78,33 +77,21 @@ class actor_addr : util::comparable<actor_addr>
intptr_t compare(const actor_addr& other) const;
intptr_t compare(const local_actor* other) const;
actor_id id() const;
const node_id& node() const;
/**
* @brief Returns whether this is an address of a
* remote actor.
*/
bool is_remote() const;
inline common_actor_ops* operator->() const {
// this const cast is safe, because common_actor_ops cannot be
// modified anyways and the offered operations are intended to
// be called on const elements
return const_cast<common_actor_ops*>(&m_ops);
}
private:
explicit actor_addr(abstract_actor*);
abstract_actor_ptr m_ptr;
common_actor_ops m_ops;
};
namespace detail {
template<typename T>
T* actor_addr_cast(const actor_addr& addr) {
return static_cast<T*>(addr.m_ptr.get());
}
} // namespace detail
} // namespace cppa
#endif // CPPA_ACTOR_ADDR_HPP
......@@ -37,7 +37,10 @@
namespace cppa {
class untyped_actor : extend<local_actor>::with<stacked> {
/**
* @extends local_actor
*/
class blocking_untyped_actor : public extend<local_actor>::with<stacked> {
};
......
......@@ -31,6 +31,7 @@
#ifndef CPPA_CHANNEL_HPP
#define CPPA_CHANNEL_HPP
#include <cstddef>
#include <type_traits>
#include "cppa/intrusive_ptr.hpp"
......@@ -41,8 +42,7 @@
namespace cppa {
class actor;
class any_tuple;
class message_header;
namespace detail { class raw_access; }
/**
* @brief Interface for all message receivers.
......@@ -51,16 +51,23 @@ class message_header;
* and is implemented by {@link actor} and {@link group}.
*/
class channel : util::comparable<channel>
, util::comparable<channel, actor> {
, util::comparable<channel, actor>
, util::comparable<channel, abstract_channel*> {
friend class detail::raw_access;
public:
channel() = default;
channel(const actor&);
channel(const std::nullptr_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) { }
channel(abstract_channel*);
channel(abstract_channel* ptr);
explicit operator bool() const;
......@@ -72,6 +79,8 @@ class channel : util::comparable<channel>
intptr_t compare(const actor& other) const;
intptr_t compare(const abstract_channel* other) const;
static intptr_t compare(const abstract_channel* lhs, const abstract_channel* rhs);
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 CPPA_COMMON_ACTOR_OPS_HPP
#define CPPA_COMMON_ACTOR_OPS_HPP
#include "cppa/abstract_actor.hpp"
namespace cppa {
class actor;
class actor_addr;
namespace detail { class raw_access; }
/**
* @brief Encapsulates actor operations that are valid for both {@link actor}
* and {@link actor_addr} handles.
*/
class common_actor_ops {
friend class actor;
friend class actor_addr;
friend class detail::raw_access;
public:
common_actor_ops() = default;
inline bool attach(attachable_ptr ptr);
/**
* @brief Convenience function that attaches the functor
* or function @p f to this actor.
*
* The actor executes <tt>f()</tt> on exit, or immediatley
* if it already finished execution.
* @param f A functor, function or lambda expression that's executed
* if the actor finishes execution.
* @returns @c true if @p f was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*/
template<typename F>
bool attach_functor(F&& f);
inline actor_id id() const;
actor_addr address() const;
const node_id& node() const;
/**
* @brief Returns whether this is an address of a
* remote actor.
*/
bool is_remote() const;
private:
common_actor_ops(abstract_actor_ptr ptr) : m_ptr(std::move(ptr)) { }
abstract_actor_ptr m_ptr;
};
template<class F>
struct functor_attachable : attachable {
F m_functor;
template<typename T>
inline functor_attachable(T&& arg) : m_functor(std::forward<T>(arg)) { }
void actor_exited(std::uint32_t reason) { m_functor(reason); }
bool matches(const attachable::token&) { return false; }
};
template<typename F>
bool common_actor_ops::attach_functor(F&& f) {
typedef typename util::rm_const_and_ref<F>::type f_type;
typedef functor_attachable<f_type> impl;
return attach(attachable_ptr{new impl(std::forward<F>(f))});
}
inline actor_id common_actor_ops::id() const {
return (m_ptr) ? m_ptr->id() : 0;
}
inline bool common_actor_ops::attach(attachable_ptr ptr) {
return m_ptr ? m_ptr->attach(std::move(ptr)) : false;
}
} // namespace cppa
#endif // CPPA_COMMON_ACTOR_OPS_HPP
......@@ -58,11 +58,13 @@
#include "cppa/local_actor.hpp"
#include "cppa/prioritizing.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/untyped_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/blocking_untyped_actor.hpp"
#include "cppa/util/type_traits.hpp"
......@@ -434,22 +436,18 @@
namespace cppa {
template<typename T, typename... Ts>
typename std::enable_if<std::is_base_of<channel, T>::value>::type
send_tuple_as(const actor& from, const intrusive_ptr<T>& to, any_tuple msg) {
to->enqueue({from.address(), to}, std::move(msg));
template<typename... Ts>
void send_tuple_as(const actor& from, const channel& to, any_tuple msg) {
to.enqueue({from->address(), to}, std::move(msg));
}
template<typename T, typename... Ts>
typename std::enable_if<std::is_base_of<channel, T>::value>::type
send_as(const actor& from, const intrusive_ptr<T>& to, Ts&&... args) {
template<typename... Ts>
void send_as(const actor& from, channel& to, Ts&&... args) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
}
void send_tuple_as(const actor& from, const actor& to, any_tuple msg);
template<typename... Ts>
void send_as(const actor& from, const actor& to, Ts&&... args) {
void send_as(const actor& from, const channel& to, Ts&&... args) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
}
/**
......@@ -632,7 +630,7 @@ namespace std {
template<>
struct hash<cppa::actor> {
inline size_t operator()(const cppa::actor& ref) const {
return static_cast<size_t>(ref.id());
return static_cast<size_t>(ref->id());
}
};
// provide convenience overlaods for aout; implemented in logging.cpp
......
......@@ -44,11 +44,13 @@ class behavior;
class any_tuple;
class self_type;
class actor_proxy;
class untyped_actor;
class abstract_actor;
class message_header;
class partial_function;
class uniform_type_info;
class primitive_variant;
class blocking_untyped_actor;
// structs
struct anything;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_RAW_ACCESS_HPP
#define CPPA_RAW_ACCESS_HPP
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/abstract_channel.hpp"
namespace cppa { namespace detail {
class raw_access {
public:
static abstract_actor* get(const actor& hdl) {
return hdl.m_ops.m_ptr.get();
}
static abstract_actor* get(const actor_addr& hdl) {
return hdl.m_ops.m_ptr.get();
}
static abstract_channel* get(const channel& hdl) {
return hdl.m_ptr.get();
}
};
// utility function to get raw access + cast to a related type in one call
template<typename T>
T* actor_addr_cast(const actor_addr& hdl) {
return static_cast<T*>(raw_access::get(hdl));
}
} } // namespace cppa::detail
#endif // CPPA_RAW_ACCESS_HPP
......@@ -386,7 +386,7 @@ class receive_policy {
&& 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);
auto ref_opt = client->sync_handler(msg_id);
// calls client->response_handle() if hdl is a dummy
// argument, forwards hdl otherwise to reply to the
// original request message
......
......@@ -42,8 +42,6 @@ struct scheduled_actor_dummy : scheduled_actor {
void quit(std::uint32_t) override;
void dequeue(behavior&) override;
void dequeue_response(behavior&, message_id) override;
void do_become(behavior&&, bool) override;
void become_waiting_for(behavior, message_id) override;
bool has_behavior() override;
scheduled_actor_type impl_type() override;
};
......
......@@ -44,6 +44,7 @@
#include "cppa/match_expr.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/message_future.hpp"
#include "cppa/message_header.hpp"
......@@ -71,7 +72,7 @@ namespace detail { class receive_policy; }
* @brief Base class for local running Actors.
* @extends actor
*/
class local_actor : public extend<abstract_actor>::with<memory_cached> {
class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_based> {
friend class detail::receive_policy;
......@@ -100,8 +101,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
message_future sync_send_tuple(const actor& dest, any_tuple what);
......@@ -115,8 +116,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
......@@ -220,7 +221,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void monitor(const actor_addr& whom);
inline void monitor(const actor& whom) {
monitor(whom.address());
monitor(whom->address());
}
/**
......@@ -230,7 +231,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void demonitor(const actor_addr& whom);
inline void demonitor(const actor& whom) {
demonitor(whom.address());
demonitor(whom->address());
}
/**
......@@ -341,13 +342,13 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
inline void mark_arrived(message_id response_id);
virtual void become_waiting_for(behavior, message_id) = 0;
//virtual void become_waiting_for(behavior, message_id) = 0;
inline detail::behavior_stack& bhvr_stack();
//inline detail::behavior_stack& bhvr_stack();
virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
//virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
inline void do_become(const behavior& bhvr, bool discard_old);
//inline void do_become(const behavior& bhvr, bool discard_old);
const char* debug_name() const;
......@@ -359,7 +360,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
protected:
inline void remove_handler(message_id id);
void remove_handler(message_id id);
void cleanup(std::uint32_t reason);
......@@ -389,7 +390,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
std::map<group_ptr, group::subscription> m_subscriptions;
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
//detail::behavior_stack m_bhvr_stack;
// set by quit
std::uint32_t m_planned_exit_reason;
......@@ -442,9 +443,9 @@ inline actor_addr& local_actor::last_sender() {
return m_current_node->sender;
}
inline void local_actor::do_unbecome() {
m_bhvr_stack.pop_async_back();
}
//inline void local_actor::do_unbecome() {
// m_bhvr_stack.pop_async_back();
//}
inline message_id local_actor::get_response_id() {
auto id = m_current_node->mid;
......@@ -466,18 +467,18 @@ inline void local_actor::mark_arrived(message_id response_id) {
if (i != last) m_pending_responses.erase(i);
}
inline detail::behavior_stack& local_actor::bhvr_stack() {
return m_bhvr_stack;
}
//inline detail::behavior_stack& local_actor::bhvr_stack() {
// return m_bhvr_stack;
//}
inline void local_actor::do_become(const behavior& bhvr, bool discard_old) {
behavior copy{bhvr};
do_become(std::move(copy), discard_old);
}
//inline void local_actor::do_become(const behavior& bhvr, bool discard_old) {
// behavior copy{bhvr};
// do_become(std::move(copy), discard_old);
//}
inline void local_actor::remove_handler(message_id id) {
m_bhvr_stack.erase(id);
}
//inline void local_actor::remove_handler(message_id id) {
// m_bhvr_stack.erase(id);
//}
inline std::uint32_t local_actor::planned_exit_reason() const {
return m_planned_exit_reason;
......
......@@ -78,9 +78,13 @@ class message_future {
message_future() = delete;
continue_helper then(const partial_function& pfun);
inline continue_helper then(const partial_function& pfun) {
continue_helper await(const partial_function& pfun);
}
inline continue_helper await(const partial_function& pfun) {
}
/**
* @brief Sets @p mexpr as event-handler for the response message.
......@@ -140,7 +144,7 @@ class message_future {
partial_function fs2bhvr(partial_function pf);
void check_consistency();
inline void check_consistency() { }
};
......
......@@ -211,7 +211,7 @@ class optional<T&> {
typedef T type;
optional() : m_value(nullptr) { }
optional(const none_t& = none) : m_value(nullptr) { }
optional(T& value) : m_value(&value) { }
......@@ -267,7 +267,7 @@ class optional<void> {
optional(const unit_t&) : m_valid(true) { }
optional(const none_t&) : m_valid(false) { }
optional(const none_t& = none) : m_valid(false) { }
inline bool valid() const { return m_valid; }
......
......@@ -69,7 +69,7 @@ class scheduled_actor;
* @brief A base class for cooperatively scheduled actors.
* @extends local_actor
*/
class scheduled_actor : public extend<local_actor>::with<mailbox_based, threadless> {
class scheduled_actor : public extend<local_actor>::with<threadless> {
typedef combined_type super;
......
......@@ -49,7 +49,7 @@
namespace cppa {
class self_type;
class untyped_actor;
class scheduled_actor;
class scheduler_helper;
class event_based_actor;
......@@ -84,7 +84,7 @@ class scheduler {
public:
typedef std::function<void(local_actor*)> init_callback;
typedef std::function<behavior(local_actor*)> actor_fun;
typedef std::function<behavior(untyped_actor*)> actor_fun;
const actor& printer() const;
......@@ -95,7 +95,7 @@ class scheduler {
* (a thread that acts as actor).
* @note Calls <tt>what->attach(...)</tt>.
*/
virtual void register_converted_context(actor* what);
virtual void register_converted_context(abstract_actor* what);
/**
* @brief Informs the scheduler about a hidden (non-actor)
......
......@@ -207,6 +207,12 @@ class stacked : public Base {
m_behavior = std::move(fun);
}
inline optional<behavior&> sync_handler(message_id msg_id) {
auto i = m_sync_handler.find(msg_id);
if (i != m_sync_handler.end()) return i->second;
return none;
}
protected:
template<typename... Ts>
......@@ -221,6 +227,8 @@ class stacked : public Base {
private:
std::map<message_id, behavior> m_sync_handler;
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
......
......@@ -155,6 +155,14 @@ class stackless : public Base {
}
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
inline optional<behavior&> sync_handler(message_id msg_id) {
return m_bhvr_stack.sync_handler(msg_id);
}
protected:
// allows actors to keep previous behaviors and enables unbecome()
......
......@@ -64,7 +64,7 @@ class scheduler_helper;
* @extends local_actor
*/
class thread_mapped_actor : public extend<local_actor, thread_mapped_actor>::
with<mailbox_based, stacked, threaded> {
with<stacked, threaded> {
typedef combined_type super;
......
......@@ -34,10 +34,14 @@
#include "cppa/extend.hpp"
#include "cppa/stackless.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
namespace cppa {
class untyped_actor : extend<local_actor>::with<stackless> {
/**
* @extends local_actor
*/
class untyped_actor : public extend<local_actor>::with<stackless> {
};
......
......@@ -46,6 +46,7 @@
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
......
......@@ -37,26 +37,14 @@
namespace cppa {
actor::actor(abstract_actor* ptr) : m_ptr(ptr) { }
actor_id actor::id() const {
return (m_ptr) ? m_ptr->id() : 0;
}
actor_addr actor::address() const {
return m_ptr ? m_ptr->address() : actor_addr{};
}
actor::actor(abstract_actor* ptr) : m_ops(ptr) { }
void actor::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
bool actor::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
if (m_ops.m_ptr) m_ops.m_ptr->enqueue(hdr, std::move(msg));
}
intptr_t actor::compare(const actor& other) const {
return channel::compare(m_ptr.get(), other.m_ptr.get());
return channel::compare(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
}
} // namespace cppa
......@@ -32,6 +32,8 @@
#include "cppa/actor_addr.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/detail/raw_access.hpp"
namespace cppa {
namespace {
......@@ -40,40 +42,30 @@ intptr_t compare_impl(const abstract_actor* lhs, const abstract_actor* rhs) {
}
} // namespace <anonymous>
actor_addr::actor_addr(const invalid_actor_addr_t&) : m_ptr(nullptr) { }
actor_addr::actor_addr(const actor& other) : m_ops(detail::raw_access::get(other)) { }
actor_addr::actor_addr(const invalid_actor_addr_t&) : m_ops(nullptr) { }
actor_addr::actor_addr(abstract_actor* ptr) : m_ptr(ptr) { }
actor_addr::actor_addr(abstract_actor* ptr) : m_ops(ptr) { }
actor_addr::operator bool() const {
return static_cast<bool>(m_ptr);
return static_cast<bool>(m_ops.m_ptr);
}
bool actor_addr::operator!() const {
return !m_ptr;
return !(m_ops.m_ptr);
}
intptr_t actor_addr::compare(const actor& other) const {
return compare_impl(m_ptr.get(), other.m_ptr.get());
return compare_impl(m_ops.m_ptr.get(), detail::raw_access::get(other));
}
intptr_t actor_addr::compare(const actor_addr& other) const {
return compare_impl(m_ptr.get(), other.m_ptr.get());
return compare_impl(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
}
intptr_t actor_addr::compare(const local_actor* other) const {
return compare_impl(m_ptr.get(), other);
}
actor_id actor_addr::id() const {
return m_ptr->id();
}
const node_id& actor_addr::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
}
bool actor_addr::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
return compare_impl(m_ops.m_ptr.get(), other);
}
} // namespace cppa
......@@ -40,6 +40,7 @@
#include "cppa/io/middleman.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/actor_registry.hpp"
namespace cppa {
......@@ -53,11 +54,11 @@ void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
}
else {
// register locally running actors to be able to deserialize them later
if (!ptr.is_remote()) {
get_actor_registry()->put(ptr.id(), detail::actor_addr_cast<abstract_actor>(ptr));
if (!ptr->is_remote()) {
get_actor_registry()->put(ptr->id(), detail::actor_addr_cast<abstract_actor>(ptr));
}
auto& pinf = ptr.node();
sink->write_value(ptr.id());
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());
}
......
......@@ -68,7 +68,7 @@ class default_broker : public broker {
: broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { }
void init() override {
enqueue({invalid_actor_addr, this}, make_any_tuple(atom("INITMSG")));
enqueue({invalid_actor_addr, channel{this}}, make_any_tuple(atom("INITMSG")));
become(
on(atom("INITMSG")) >> [=] {
unbecome();
......@@ -347,7 +347,9 @@ bool broker::initialized() const {
}
void broker::quit(std::uint32_t reason) {
cleanup(reason);
//cleanup(reason);
m_bhvr_stack.clear();
super::quit(reason);
}
void broker::init_broker() {
......
......@@ -32,8 +32,14 @@
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/detail/raw_access.hpp"
namespace cppa {
channel::channel(const std::nullptr_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) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
......@@ -57,7 +63,11 @@ intptr_t channel::compare(const channel& other) const {
}
intptr_t channel::compare(const actor& other) const {
return compare(m_ptr.get(), other.m_ptr.get());
return compare(m_ptr.get(), detail::raw_access::get(other));
}
intptr_t channel::compare(const abstract_channel* other) const {
return compare(m_ptr.get(), other);
}
} // namespace cppa
\ No newline at end of file
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/actor_addr.hpp"
#include "cppa/common_actor_ops.hpp"
namespace cppa {
actor_addr common_actor_ops::address() const {
return m_ptr ? m_ptr->address() : actor_addr{};
}
const node_id& common_actor_ops::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
}
bool common_actor_ops::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
}
} // namespace cppa
\ No newline at end of file
......@@ -40,12 +40,15 @@
#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/untyped_actor.hpp"
#include "cppa/message_header.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/message_header.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......@@ -308,7 +311,7 @@ class local_group_module : public group::module {
actor broker;
m_actor_utype->deserialize(&broker, source);
if (!broker) return nullptr;
if (!broker.is_remote()) {
if (!broker->is_remote()) {
return this->get(identifier);
}
else {
......@@ -454,7 +457,7 @@ class remote_group_module : public group::module {
auto sm = make_counted<shared_map>();
group::module_ptr _this = this;
m_map = sm;
m_map->m_worker = spawn<hidden>([=](event_based_actor* self) -> behavior {
m_map->m_worker = spawn<hidden>([=](untyped_actor* self) -> behavior {
CPPA_LOGC_TRACE(detail::demangle(typeid(*_this)),
"remote_group_module$worker",
"");
......
......@@ -36,6 +36,8 @@
#include "cppa/local_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/detail/raw_access.hpp"
namespace cppa {
namespace {
......@@ -160,7 +162,7 @@ void local_actor::reply_message(any_tuple&& what) {
void local_actor::forward_message(const actor& dest, message_priority p) {
if (!dest) return;
auto& id = m_current_node->mid;
dest.m_ptr->enqueue({m_current_node->sender, dest.m_ptr, id, p}, m_current_node->msg);
detail::raw_access::get(dest)->enqueue({m_current_node->sender, detail::raw_access::get(dest), id, p}, m_current_node->msg);
// treat this message as asynchronous message from now on
id = message_id{};
}
......@@ -174,6 +176,28 @@ response_handle local_actor::make_response_handle() {
}
*/
void local_actor::send_tuple(const channel& dest, any_tuple what) {
//TODO:
}
void local_actor::remove_handler(message_id) {
}
void local_actor::delayed_send_tuple(const channel&, const util::duration&, cppa::any_tuple) {
}
message_future local_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
}
response_handle local_actor::make_response_handle() {
return {};
}
void local_actor::cleanup(std::uint32_t reason) {
m_subscriptions.clear();
super::cleanup(reason);
......@@ -193,7 +217,6 @@ void local_actor::quit(std::uint32_t reason) {
if (reason == exit_reason::unallowed_function_call) {
// this is the only reason that causes an exception
cleanup(reason);
m_bhvr_stack.clear();
CPPA_LOG_WARNING("actor tried to use a blocking function");
// when using receive(), the non-blocking nature of event-based
// actors breaks any assumption the user has about his code,
......@@ -203,7 +226,6 @@ void local_actor::quit(std::uint32_t reason) {
"use receive()\n";
throw actor_exited(reason);
}
m_bhvr_stack.clear();
planned_exit_reason(reason);
}
......
......@@ -47,6 +47,7 @@
#include "cppa/util/algorithm.hpp"
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
......@@ -218,7 +219,7 @@ void peer::monitor(const actor_addr&,
else {
CPPA_LOGMF(CPPA_DEBUG, self, "attach functor to " << entry.first.get());
auto mm = parent();
entry.first->attach_functor([=](uint32_t reason) {
entry.first->address()->attach_functor([=](uint32_t reason) {
mm->run_later([=] {
CPPA_LOGC_TRACE("cppa::io::peer",
"monitor$kill_proxy_helper",
......@@ -261,7 +262,7 @@ void peer::kill_proxy(const actor_addr& sender,
void peer::deliver(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE("");
if (hdr.sender && hdr.sender.is_remote()) {
if (hdr.sender && hdr.sender->is_remote()) {
auto ptr = detail::actor_addr_cast<actor_proxy>(hdr.sender);
ptr->deliver(hdr, std::move(msg));
}
......@@ -280,7 +281,7 @@ void peer::link(const actor_addr& sender, const actor_addr& receiver) {
auto locally_link_proxy = [](const actor_addr& lhs, const actor_addr& rhs) {
detail::actor_addr_cast<actor_proxy>(lhs)->local_link_to(rhs);
};
switch ((sender.is_remote() ? 0x10 : 0x00) | (receiver.is_remote() ? 0x01 : 0x00)) {
switch ((sender->is_remote() ? 0x10 : 0x00) | (receiver->is_remote() ? 0x01 : 0x00)) {
case 0x00: // both local
case 0x11: // both remote
detail::actor_addr_cast<abstract_actor>(sender)->link_to(receiver);
......@@ -304,7 +305,7 @@ void peer::unlink(const actor_addr& sender, const actor_addr& receiver) {
auto locally_unlink_proxy = [](const actor_addr& lhs, const actor_addr& rhs) {
detail::actor_addr_cast<actor_proxy>(lhs)->local_unlink_from(rhs);
};
switch ((sender.is_remote() ? 0x10 : 0x00) | (receiver.is_remote() ? 0x01 : 0x00)) {
switch ((sender->is_remote() ? 0x10 : 0x00) | (receiver->is_remote() ? 0x01 : 0x00)) {
case 0x00: // both local
case 0x11: // both remote
detail::actor_addr_cast<abstract_actor>(sender)->unlink_from(receiver);
......
......@@ -64,7 +64,7 @@ continue_reading_result peer_acceptor::continue_reading() {
auto& pself = node_id::get();
uint32_t process_id = pself->process_id();
try {
actor_id aid = published_actor().id();
actor_id aid = published_actor()->id();
pair.second->write(&aid, sizeof(actor_id));
pair.second->write(&process_id, sizeof(uint32_t));
pair.second->write(pself->host_id().data(),
......
......@@ -33,6 +33,8 @@
#include "cppa/local_actor.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/detail/raw_access.hpp"
using std::move;
namespace cppa {
......
......@@ -39,8 +39,6 @@ void scheduled_actor_dummy::enqueue(const message_header&, any_tuple) { }
void scheduled_actor_dummy::quit(std::uint32_t) { }
void scheduled_actor_dummy::dequeue(behavior&) { }
void scheduled_actor_dummy::dequeue_response(behavior&, message_id) { }
void scheduled_actor_dummy::do_become(behavior&&, bool) { }
void scheduled_actor_dummy::become_waiting_for(behavior, message_id) { }
bool scheduled_actor_dummy::has_behavior() { return false; }
resume_result scheduled_actor_dummy::resume(util::fiber*) {
......
......@@ -96,8 +96,8 @@ class scheduler_helper {
auto mt = make_counted<thread_mapped_actor>();
auto mp = make_counted<thread_mapped_actor>();
// launch threads
m_timer_thread = std::thread{&scheduler_helper::timer_loop, mt};
m_printer_thread = std::thread{&scheduler_helper::printer_loop, mp};
m_timer_thread = std::thread{&scheduler_helper::timer_loop, mt.get()};
m_printer_thread = std::thread{&scheduler_helper::printer_loop, mp.get()};
// set member variables
m_timer = std::move(mt);
m_printer = std::move(mp);
......@@ -185,8 +185,8 @@ void scheduler_helper::timer_loop(thread_mapped_actor* self) {
}
void scheduler_helper::printer_loop(thread_mapped_actor* self) {
std::map<actor, std::string> out;
auto flush_output = [&out](const actor& s) {
std::map<actor_addr, std::string> out;
auto flush_output = [&out](const actor_addr& s) {
auto i = out.find(s);
if (i != out.end()) {
auto& line = i->second;
......@@ -261,10 +261,10 @@ actor& scheduler::delayed_send_helper() {
return m_helper->m_timer;
}
void scheduler::register_converted_context(actor* what) {
void scheduler::register_converted_context(abstract_actor* what) {
if (what) {
get_actor_registry()->inc_running();
what->attach(attachable_ptr{new exit_observer});
what->address()->attach(attachable_ptr{new exit_observer});
}
}
......
......@@ -33,6 +33,8 @@
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
namespace cppa { namespace detail {
......@@ -44,7 +46,7 @@ void sync_request_bouncer::operator()(const actor_addr& sender, const message_id
CPPA_REQUIRE(rsn != exit_reason::not_exited);
if (sender && mid.is_request()) {
auto ptr = detail::actor_addr_cast<abstract_actor>(sender);
ptr->enqueue({nullptr, ptr, mid.response_id(),
ptr->enqueue({invalid_actor_addr, ptr, mid.response_id()},
make_any_tuple(atom("EXITED"), rsn));
}
}
......
......@@ -62,7 +62,7 @@ struct thread_pool_scheduler::worker {
worker(job_queue* jq, job_ptr dummy) : m_job_queue(jq), m_dummy(dummy) { }
void start() {
m_thread = std::thread(&thread_pool_scheduler::worker_loop, this);
//m_thread = std::thread(&thread_pool_scheduler::worker_loop, this);
}
worker(const worker&) = delete;
......@@ -179,6 +179,7 @@ void thread_pool_scheduler::enqueue(scheduled_actor* what) {
}
void exec_as_thread(bool is_hidden, intrusive_ptr<untyped_actor> ptr) {
/*TODO:
if (!is_hidden) get_actor_registry()->inc_running();
std::thread([=] {
try { ptr->exec_bhvr_stack(); }
......@@ -188,13 +189,15 @@ void exec_as_thread(bool is_hidden, intrusive_ptr<untyped_actor> ptr) {
get_actor_registry()->dec_running();
}
}).detach();
*/
}
local_actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
/*TODO:
CPPA_REQUIRE(p != nullptr);
bool is_hidden = has_hide_flag(os);
if (has_detach_flag(os)) {
exec_as_thread(is_hidden, [p] {
exec_as_thread(is_hidden, p [p] {
p->run_detached();
});
return p;
......@@ -207,6 +210,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_pt
}
else p->on_exit();
return p;
*/
}
local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
......@@ -221,6 +225,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
if (has_priority_aware_flag(os)) {
using impl = extend<local_actor>::with<stackless, threaded, prioritizing>;
set_result(make_counted<impl>());
/*TODO:
exec_as_thread(has_hide_flag(os), [result, f] {
try {
f();
......@@ -238,6 +243,7 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
}
result->on_exit();
});
*/
}
else if (has_blocking_api_flag(os)) {
# ifndef CPPA_DISABLE_CONTEXT_SWITCHING
......@@ -249,18 +255,22 @@ local_actor_ptr thread_pool_scheduler::exec(spawn_options os,
else
# endif
/* else tree */ {
auto p = make_counted<thread_mapped_actor>(std::move(f));
set_result(p);
//auto p = make_counted<thread_mapped_actor>(std::move(f));
//set_result(p);
/*TODO:
exec_as_thread(has_hide_flag(os), [p] {
p->run();
p->on_exit();
});
*/
}
}
else {
/*TODO:
auto p = event_based_actor::from(std::move(f));
set_result(p);
exec(os, p);
*/
}
CPPA_REQUIRE(result != nullptr);
return result;
......
......@@ -48,6 +48,8 @@
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/intrusive/blocking_single_reader_queue.hpp"
......@@ -68,7 +70,7 @@ void publish(actor whom, std::unique_ptr<acceptor> aptr) {
<< ", args.size() = " << args.size());
if (!whom) return;
CPPA_REQUIRE(args.size() == 0);
get_actor_registry()->put(whom.id(), detail::actor_addr_cast<abstract_actor>(whom));
get_actor_registry()->put(whom->id(), detail::actor_addr_cast<abstract_actor>(whom));
auto mm = get_middleman();
mm->register_acceptor(whom, new peer_acceptor(mm, move(aptr), whom));
}
......
......@@ -48,6 +48,7 @@
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
......@@ -179,7 +180,7 @@ void deserialize_impl(actor_addr& addr, deserializer* source) {
}
void serialize_impl(const actor& ptr, serializer* sink) {
serialize_impl(ptr.address(), sink);
serialize_impl(ptr->address(), sink);
}
void deserialize_impl(actor& ptr, deserializer* source) {
......@@ -218,14 +219,15 @@ void serialize_impl(const channel& ptr, serializer* sink) {
};
if (ptr == nullptr) wr_nullptr();
else {
auto aptr = ptr.downcast<abstract_actor>();
auto rptr = detail::raw_access::get(ptr);
auto aptr = dynamic_cast<abstract_actor*>(rptr);
if (aptr != nullptr) {
flag = 1;
sink->write_value(flag);
serialize_impl(actor{aptr}, sink);
}
else {
auto gptr = ptr.downcast<group>();
auto gptr = group_ptr{dynamic_cast<group*>(rptr)};
if (gptr != nullptr) {
flag = 2;
sink->write_value(flag);
......@@ -243,7 +245,7 @@ void deserialize_impl(channel& ptrref, deserializer* source) {
auto flag = source->read<uint8_t>();
switch (flag) {
case 0: {
ptrref.reset();
ptrref = channel{}; //ptrref.reset();
break;
}
case 1: {
......
......@@ -6,13 +6,13 @@
#include <cstddef>
#include "cppa/cppa_fwd.hpp"
void ping(size_t num_pings);
void ping(cppa::blocking_untyped_actor*, size_t num_pings);
void event_based_ping(size_t num_pings);
void event_based_ping(cppa::untyped_actor*, size_t num_pings);
void pong(cppa::actor_ptr ping_actor);
void pong(cppa::blocking_untyped_actor*, cppa::actor_ptr ping_actor);
void event_based_pong(cppa::actor_ptr ping_actor);
void event_based_pong(cppa::untyped_actor*, cppa::actor_ptr ping_actor);
// returns the number of messages ping received
size_t pongs();
......
......@@ -34,8 +34,7 @@ const char* cppa_strip_path(const char* fname) {
}
void cppa_unexpected_message(const char* fname, size_t line_num) {
CPPA_PRINTERRC(fname, line_num,
"unexpected message: " << to_string(self->last_dequeued()));
CPPA_PRINTERRC(fname, line_num, "unexpected message");
}
void cppa_unexpected_timeout(const char* fname, size_t line_num) {
......
......@@ -56,7 +56,7 @@ const T& cppa_stream_arg(const T& value) {
return value;
}
inline std::string cppa_stream_arg(const cppa::actor_ptr& ptr) {
inline std::string cppa_stream_arg(const cppa::actor& ptr) {
return cppa::to_string(ptr);
}
......@@ -178,7 +178,7 @@ void run_client_part(const std::map<std::string, std::string>& args, F fun) {
}
auto port = static_cast<std::uint16_t>(stoi(i->second));
fun(port);
cppa::await_all_others_done();
cppa::await_all_actors_done();
cppa::shutdown();
}
......
......@@ -39,12 +39,12 @@ using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void ping(size_t num_pings) {
void ping(cppa::untyped_actor* self, size_t num_pings) {
auto count = std::make_shared<size_t>(0);
become (
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) {
send(pong, atom("ping"), 1);
become (
self->become (
on(atom("kickoff"), arg_match) >> [=](const actor& pong) {
self->send(pong, atom("ping"), 1);
self->become (
on(atom("pong"), arg_match)
>> [=](int value) -> cow_tuple<atom_value, int> {
if (++*count >= num_pings) self->quit();
......@@ -57,13 +57,13 @@ void ping(size_t num_pings) {
);
}
void pong() {
become (
void pong(cppa::untyped_actor* self) {
self->become (
on(atom("ping"), arg_match)
>> [](int value) -> cow_tuple<atom_value, int> {
>> [=](int value) -> cow_tuple<atom_value, int> {
self->monitor(self->last_sender());
// set next behavior
become (
self->become (
on(atom("ping"), arg_match) >> [](int value) {
return make_cow_tuple(atom("pong"), value);
},
......@@ -79,17 +79,17 @@ void pong() {
);
}
void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy) {
void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
self->monitor(buddy);
if (thisptr->num_connections() == 0) {
if (self->num_connections() == 0) {
cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1");
}
auto write = [=](atom_value type, int value) {
thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value);
self->write(hdl, sizeof(type), &type);
self->write(hdl, sizeof(value), &value);
};
become (
self->become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
self->quit();
},
......@@ -98,7 +98,7 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
int value;
memcpy(&type, buf.data(), sizeof(atom_value));
memcpy(&value, buf.offset_data(sizeof(atom_value)), sizeof(int));
send(buddy, type, value);
self->send(buddy, type, value);
},
on(atom("ping"), arg_match) >> [=](int value) {
write(atom("ping"), value);
......@@ -107,18 +107,18 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
write(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (thisptr->last_sender() == buddy) self->quit(reason);
if (self->last_sender() == buddy) self->quit(reason);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
become (
void peer_acceptor(io::broker* self, const actor& buddy) {
self->become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) {
CPPA_CHECKPOINT();
CPPA_LOGF_INFO("received IO_accept");
thisptr->fork(peer, hdl, buddy);
self->fork(peer, hdl, buddy);
self->quit();
},
others() >> CPPA_UNEXPECTED_MSG_CB()
......
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