Commit 6773bedb authored by Dominik Charousset's avatar Dominik Charousset

preparations for 0.9 changes

this commit is the first part of a revision of libcppa that removes
the thread-local `self` pointer, `actor_ptr`, and `channel_ptr`
to pave the way for fully type-safe actor programming;
this first step splits the addressing of actors into an `actor`
handle and an `actor_addr`: only the former can be used to send
messages, whereas the latter can only be used to monitor or identify
actors;
next steps will add `typed_actor<>` handles that allow the compiler
to type-check the message passing interface;
the revision is work in progress and this commit does not compile
parent d098bcb5
......@@ -120,8 +120,11 @@ file(GLOB LIBCPPA_HDRS "cppa/*.hpp"
# list cpp files including platform-dependent files
set(LIBCPPA_SRC
${LIBCPPA_PLATFORM_SRC}
src/abstract_actor.cpp
src/abstract_channel.cpp
src/abstract_tuple.cpp
src/actor.cpp
src/actor_addr.cpp
src/actor_namespace.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
......@@ -149,7 +152,6 @@ set(LIBCPPA_SRC
src/event_based_actor.cpp
src/exception.cpp
src/exit_reason.cpp
src/factory.cpp
src/fd_util.cpp
src/fiber.cpp
src/get_root_uuid.cpp
......@@ -181,8 +183,7 @@ set(LIBCPPA_SRC
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp
src/send.cpp
src/self.cpp
src/scoped_actor.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_ABSTRACT_ACTOR_HPP
#define CPPA_ABSTRACT_ACTOR_HPP
#include <mutex>
#include <atomic>
#include <memory>
#include <vector>
#include <cstdint>
#include <type_traits>
#include "cppa/node_id.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/attachable.hpp"
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_channel.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
class actor_addr;
class serializer;
class deserializer;
/**
* @brief A unique actor ID.
* @relates actor
*/
typedef std::uint32_t actor_id;
class abstract_actor;
typedef intrusive_ptr<abstract_actor> abstract_actor_ptr;
/**
* @brief Base class for all actor implementations.
*/
class abstract_actor : public abstract_channel {
public:
/**
* @brief Attaches @p ptr to this actor.
*
* The actor will call <tt>ptr->detach(...)</tt> on exit, or immediately
* if it already finished execution.
* @param ptr A callback object that's executed if the actor finishes
* execution.
* @returns @c true if @p ptr was successfully attached to the actor;
* otherwise (actor already exited) @p false.
* @warning The actor takes ownership of @p ptr.
*/
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.
*/
actor_addr address() const;
/**
* @brief Detaches the first attached object that matches @p what.
*/
void detach(const attachable::token& what);
/**
* @brief Links this actor to @p other.
* @param other Actor instance that whose execution is coupled to the
* execution of this Actor.
*/
virtual void link_to(const actor_addr& other);
/**
* @brief Unlinks this actor from @p other.
* @param other Linked Actor.
* @note Links are automatically removed if the actor finishes execution.
*/
virtual void unlink_from(const actor_addr& other);
/**
* @brief Establishes a link relation between this actor and @p other.
* @param other Actor instance that wants to link against this Actor.
* @returns @c true if this actor is running and added @p other to its
* list of linked actors; otherwise @c false.
*/
virtual bool establish_backlink(const actor_addr& other);
/**
* @brief Removes a link relation between this actor and @p other.
* @param other Actor instance that wants to unlink from this Actor.
* @returns @c true if this actor is running and removed @p other
* from its list of linked actors; otherwise @c false.
*/
virtual bool remove_backlink(const actor_addr& other);
/**
* @brief Gets an integer value that uniquely identifies this Actor in
* the process it's executed in.
* @returns The unique identifier of this actor.
*/
inline actor_id id() const;
/**
* @brief Checks if this actor is running on a remote node.
* @returns @c true if this actor represents a remote actor;
* otherwise @c false.
*/
inline bool is_proxy() const;
/**
* @brief Returns the ID of the node this actor is running on.
*/
inline const node_id& node() const;
protected:
abstract_actor();
abstract_actor(actor_id aid);
/**
* @brief Should be overridden by subtypes and called upon termination.
* @note Default implementation sets 'exit_reason' accordingly.
* @note Overridden functions should always call super::cleanup().
*/
virtual void cleanup(std::uint32_t reason);
/**
* @brief The default implementation for {@link link_to()}.
*/
bool link_to_impl(const actor_addr& other);
/**
* @brief The default implementation for {@link unlink_from()}.
*/
bool unlink_from_impl(const actor_addr& other);
/**
* @brief Returns the actor's exit reason of
* <tt>exit_reason::not_exited</tt> if it's still alive.
*/
inline std::uint32_t exit_reason() const;
/**
* @brief Returns <tt>exit_reason() != exit_reason::not_exited</tt>.
*/
inline bool exited() const;
// cannot be changed after construction
const actor_id m_id;
// you're either a proxy or you're not
const bool m_is_proxy;
private:
// initially exit_reason::not_exited
std::atomic<std::uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, and m_links
std::mutex m_mtx;
// links to other actors
std::vector<abstract_actor_ptr> m_links;
// attached functors that are executed on cleanup
std::vector<attachable_ptr> m_attachables;
// identifies the node this actor is running on
node_id_ptr m_node;
};
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline std::uint32_t abstract_actor::id() const {
return m_id;
}
inline bool abstract_actor::is_proxy() const {
return m_is_proxy;
}
inline std::uint32_t abstract_actor::exit_reason() const {
return m_exit_reason;
}
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;
}
} // namespace cppa
#endif // CPPA_ABSTRACT_ACTOR_HPP
......@@ -28,52 +28,38 @@
\******************************************************************************/
#ifndef CPPA_FACTORY_HPP
#define CPPA_FACTORY_HPP
#ifndef CPPA_ABSTRACT_CHANNEL_HPP
#define CPPA_ABSTRACT_CHANNEL_HPP
#include "cppa/detail/event_based_actor_factory.hpp"
#include "cppa/ref_counted.hpp"
namespace cppa { namespace factory {
namespace cppa {
void default_cleanup();
// forward declarations
class any_tuple;
class message_header;
/**
* @brief Returns a factory for event-based actors using @p fun as
* implementation for {@link cppa::event_based_actor::init() init()}.
* @brief Interface for all message receivers.
*
* @p fun must take pointer arguments only. The factory creates an event-based
* actor implementation with member variables according to the functor's
* signature, as shown in the example below.
*
* @code
* auto f = factory::event_based([](int* a, int* b) { ... });
* auto actor1 = f.spawn();
* auto actor2 = f.spawn(1);
* auto actor3 = f.spawn(1, 2);
* @endcode
*
* The arguments @p a and @p b will point to @p int member variables of the
* actor. All member variables are initialized using the default constructor
* unless an initial value is passed to @p spawn.
* This interface describes an entity that can receive messages
* and is implemented by {@link actor} and {@link group}.
*/
template<typename InitFun>
inline typename detail::ebaf_from_functor<InitFun, void (*)()>::type
event_based(InitFun init) {
return {std::move(init), default_cleanup};
}
class abstract_channel : public ref_counted {
/**
* @brief Returns a factory for event-based actors using @p fun0 as
* implementation for {@link cppa::event_based_actor::init() init()}
* and @p fun1 as implementation for
* {@link cppa::event_based_actor::on_exit() on_exit()}.
*/
template<typename InitFun, typename OnExitFun>
inline typename detail::ebaf_from_functor<InitFun, OnExitFun>::type
event_based(InitFun init, OnExitFun on_exit) {
return {std::move(init), on_exit};
}
public:
/**
* @brief Enqueues @p msg to the list of received messages.
*/
virtual void enqueue(const message_header& hdr, any_tuple msg) = 0;
protected:
virtual ~abstract_channel();
};
} } // namespace cppa::factory
} // namespace cppa
#endif // CPPA_FACTORY_HPP
#endif // CPPA_ABSTRACT_CHANNEL_HPP
......@@ -31,227 +31,60 @@
#ifndef CPPA_ACTOR_HPP
#define CPPA_ACTOR_HPP
#include <mutex>
#include <atomic>
#include <memory>
#include <vector>
#include <cstdint>
#include <type_traits>
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/attachable.hpp"
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/util/type_traits.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
class actor;
class self_type;
class serializer;
class deserializer;
class channel;
class any_tuple;
class actor_addr;
class local_actor;
class message_header;
/**
* @brief A unique actor ID.
* @relates actor
* @brief Identifies an untyped actor.
*/
typedef std::uint32_t actor_id;
class actor : util::comparable<actor> {
/**
* @brief A smart pointer type that manages instances of {@link actor}.
* @relates actor
*/
typedef intrusive_ptr<actor> actor_ptr;
/**
* @brief Base class for all actor implementations.
*/
class actor : public channel {
friend class channel;
friend class actor_addr;
friend class local_actor;
public:
/**
* @brief Attaches @p ptr to this actor.
*
* The actor will call <tt>ptr->detach(...)</tt> on exit, or immediately
* if it already finished execution.
* @param ptr A callback object that's executed if the actor finishes
* execution.
* @returns @c true if @p ptr was successfully attached to the actor;
* otherwise (actor already exited) @p false.
* @warning The actor takes ownership of @p ptr.
*/
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 Detaches the first attached object that matches @p what.
*/
void detach(const attachable::token& what);
/**
* @brief Links this actor to @p other.
* @param other Actor instance that whose execution is coupled to the
* execution of this Actor.
*/
virtual void link_to(const actor_ptr& other);
/**
* @brief Unlinks this actor from @p other.
* @param other Linked Actor.
* @note Links are automatically removed if the actor finishes execution.
*/
virtual void unlink_from(const actor_ptr& other);
actor() = default;
/**
* @brief Establishes a link relation between this actor and @p other.
* @param other Actor instance that wants to link against this Actor.
* @returns @c true if this actor is running and added @p other to its
* list of linked actors; otherwise @c false.
*/
virtual bool establish_backlink(const actor_ptr& other);
/**
* @brief Removes a link relation between this actor and @p other.
* @param other Actor instance that wants to unlink from this Actor.
* @returns @c true if this actor is running and removed @p other
* from its list of linked actors; otherwise @c false.
*/
virtual bool remove_backlink(const actor_ptr& other);
/**
* @brief Gets an integer value that uniquely identifies this Actor in
* the process it's executed in.
* @returns The unique identifier of this actor.
*/
inline actor_id id() const;
/**
* @brief Checks if this actor is running on a remote node.
* @returns @c true if this actor represents a remote actor;
* otherwise @c false.
*/
inline bool is_proxy() const;
protected:
actor();
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(actor_id aid);
actor(abstract_actor*);
/**
* @brief Should be overridden by subtypes and called upon termination.
* @note Default implementation sets 'exit_reason' accordingly.
* @note Overridden functions should always call super::cleanup().
*/
virtual void cleanup(std::uint32_t reason);
actor_id id() const;
/**
* @brief The default implementation for {@link link_to()}.
*/
bool link_to_impl(const actor_ptr& other);
actor_addr address() const;
/**
* @brief The default implementation for {@link unlink_from()}.
*/
bool unlink_from_impl(const actor_ptr& other);
explicit operator bool() const;
/**
* @brief Returns the actor's exit reason of
* <tt>exit_reason::not_exited</tt> if it's still alive.
*/
inline std::uint32_t exit_reason() const;
bool operator!() const;
/**
* @brief Returns <tt>exit_reason() != exit_reason::not_exited</tt>.
*/
inline bool exited() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
// cannot be changed after construction
const actor_id m_id;
bool is_remote() const;
// you're either a proxy or you're not
const bool m_is_proxy;
intptr_t compare(const actor& other) const;
private:
// initially exit_reason::not_exited
std::atomic<std::uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, and m_links
std::mutex m_mtx;
// links to other actors
std::vector<actor_ptr> m_links;
// attached functors that are executed on cleanup
std::vector<attachable_ptr> m_attachables;
intrusive_ptr<abstract_actor> m_ptr;
};
// undocumented, because self_type is hidden in documentation
bool operator==(const actor_ptr&, const self_type&);
bool operator!=(const self_type&, const actor_ptr&);
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline std::uint32_t actor::id() const {
return m_id;
}
inline bool actor::is_proxy() const {
return m_is_proxy;
}
inline std::uint32_t actor::exit_reason() const {
return m_exit_reason;
}
inline bool 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 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))});
}
} // namespace cppa
#endif // CPPA_ACTOR_HPP
......@@ -25,135 +25,86 @@
* *
* 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_SELF_HPP
#define CPPA_SELF_HPP
#ifndef CPPA_ACTOR_ADDR_HPP
#define CPPA_ACTOR_ADDR_HPP
#include <cstddef>
#include <cstdint>
#include <type_traits>
#include "cppa/actor.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @brief Always points to the current actor. Similar to @c this in
* an object-oriented context.
*/
extern local_actor* self;
#else // CPPA_DOCUMENTATION
class actor;
class node_id;
class actor_addr;
class local_actor;
// convertible<...> enables "actor_ptr this_actor = self;"
class self_type : public convertible<self_type, actor*>,
public convertible<self_type, channel*> {
namespace detail { template<typename T> T* actor_addr_cast(const actor_addr&); }
self_type(const self_type&) = delete;
self_type& operator=(const self_type&) = delete;
constexpr struct invalid_actor_addr_t { constexpr invalid_actor_addr_t() { } } invalid_actor_addr;
public:
class actor_addr : util::comparable<actor_addr>
, util::comparable<actor_addr, actor>
, util::comparable<actor_addr, local_actor*> {
typedef local_actor* pointer;
friend class abstract_actor;
constexpr self_type() { }
template<typename T>
friend T* detail::actor_addr_cast(const actor_addr&);
// "inherited" from convertible<...>
inline actor* do_convert() const {
return convert_impl();
}
public:
inline pointer get() const {
return get_impl();
}
actor_addr() = default;
actor_addr(actor_addr&&) = default;
actor_addr(const actor_addr&) = default;
actor_addr& operator=(actor_addr&&) = default;
actor_addr& operator=(const actor_addr&) = default;
// allow "self" wherever an local_actor or actor pointer is expected
inline operator pointer() const {
return get_impl();
}
actor_addr(const actor&);
inline pointer operator->() const {
return get_impl();
}
actor_addr(const invalid_actor_addr_t&);
// @pre get_unchecked() == nullptr
inline void set(pointer ptr) const {
set_impl(ptr);
}
explicit operator bool() const;
bool operator!() const;
// @returns The current value without converting the calling context
// to an actor on-the-fly.
inline pointer unchecked() const {
return get_unchecked_impl();
}
intptr_t compare(const actor& other) const;
intptr_t compare(const actor_addr& other) const;
intptr_t compare(const local_actor* other) const;
inline pointer release() const {
return release_impl();
}
actor_id id() const;
inline void adopt(pointer ptr) const {
adopt_impl(ptr);
}
const node_id& node() const;
static void cleanup_fun(pointer);
/**
* @brief Returns whether this is an address of a
* remote actor.
*/
bool is_remote() const;
private:
static void set_impl(pointer);
explicit actor_addr(abstract_actor*);
static pointer get_unchecked_impl();
static pointer get_impl();
static actor* convert_impl();
static pointer release_impl();
static void adopt_impl(pointer);
abstract_actor_ptr m_ptr;
};
/*
* "self" emulates a new keyword. The object itself is useless, all it does
* is to provide syntactic sugar like "self->trap_exit(true)".
*/
constexpr self_type self;
class scoped_self_setter {
scoped_self_setter(const scoped_self_setter&) = delete;
scoped_self_setter& operator=(const scoped_self_setter&) = delete;
public:
inline scoped_self_setter(local_actor* new_value) {
m_original_value = self.release();
self.adopt(new_value);
}
inline ~scoped_self_setter() {
// restore self
static_cast<void>(self.release());
self.adopt(m_original_value);
}
private:
local_actor* m_original_value;
};
namespace detail {
// disambiguation (compiler gets confused with cast operator otherwise)
bool operator==(const actor_ptr& lhs, const self_type& rhs);
bool operator==(const self_type& lhs, const actor_ptr& rhs);
bool operator!=(const actor_ptr& lhs, const self_type& rhs);
bool operator!=(const self_type& lhs, const actor_ptr& rhs);
template<typename T>
T* actor_addr_cast(const actor_addr& addr) {
return static_cast<T*>(addr.m_ptr.get());
}
#endif // CPPA_DOCUMENTATION
} // namespace detail
} // namespace cppa
#endif // CPPA_SELF_HPP
#endif // CPPA_ACTOR_ADDR_HPP
......@@ -63,9 +63,9 @@ class actor_namespace {
inline void set_new_element_callback(new_element_callback fun);
void write(serializer* sink, const actor_ptr& ptr);
void write(serializer* sink, const actor_addr& ptr);
actor_ptr read(deserializer* source);
actor_addr read(deserializer* source);
/**
* @brief A map that stores weak actor proxy pointers by actor ids.
......@@ -81,13 +81,13 @@ class actor_namespace {
* @brief Returns the proxy instance identified by @p node and @p aid
* or @p nullptr if the actor is unknown.
*/
actor_ptr get(const node_id& node, actor_id aid);
actor_proxy_ptr get(const node_id& node, actor_id aid);
/**
* @brief Returns the proxy instance identified by @p node and @p aid
* or creates a new (default) proxy instance.
*/
actor_ptr get_or_put(node_id_ptr node, actor_id aid);
actor_proxy_ptr get_or_put(node_id_ptr node, actor_id aid);
/**
* @brief Stores @p proxy in the list of known actor proxies.
......
......@@ -31,8 +31,9 @@
#ifndef CPPA_ACTOR_PROXY_HPP
#define CPPA_ACTOR_PROXY_HPP
#include "cppa/actor.hpp"
#include "cppa/extend.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/message_header.hpp"
#include "cppa/enable_weak_ptr.hpp"
#include "cppa/weak_intrusive_ptr.hpp"
......@@ -45,7 +46,7 @@ class actor_proxy_cache;
* @brief Represents a remote actor.
* @extends actor
*/
class actor_proxy : public extend<actor>::with<enable_weak_ptr> {
class actor_proxy : public extend<abstract_actor>::with<enable_weak_ptr> {
typedef combined_type super;
......@@ -55,12 +56,12 @@ class actor_proxy : public extend<actor>::with<enable_weak_ptr> {
* @brief Establishes a local link state that's not synchronized back
* to the remote instance.
*/
virtual void local_link_to(const intrusive_ptr<actor>& other) = 0;
virtual void local_link_to(const actor_addr& other) = 0;
/**
* @brief Removes a local link state.
*/
virtual void local_unlink_from(const actor_ptr& other) = 0;
virtual void local_unlink_from(const actor_addr& other) = 0;
/**
* @brief Delivers given message via this proxy instance.
......
......@@ -91,7 +91,7 @@ class any_tuple {
* @brief Creates a tuple from a set of values.
*/
template<typename T, typename... Ts>
any_tuple(T v0, Ts&&... vs)
explicit any_tuple(T v0, Ts&&... vs)
: m_vals(make_cow_tuple(std::move(v0), std::forward<Ts>(vs)...).vals()) { }
/**
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_BLOCKING_UNTYPED_ACTOR_HPP
#define CPPA_BLOCKING_UNTYPED_ACTOR_HPP
#include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
class untyped_actor : extend<local_actor>::with<stacked> {
};
} // namespace cppa
#endif // CPPA_BLOCKING_UNTYPED_ACTOR_HPP
......@@ -25,7 +25,7 @@
* *
* 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_CHANNEL_HPP
......@@ -33,68 +33,53 @@
#include <type_traits>
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_channel.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
// forward declarations
class actor;
class group;
class any_tuple;
class message_header;
typedef intrusive_ptr<actor> actor_ptr;
/**
* @brief Interface for all message receivers.
*
* This interface describes an entity that can receive messages
* and is implemented by {@link actor} and {@link group}.
*/
class channel : public ref_counted {
friend class actor;
friend class group;
class channel : util::comparable<channel>
, util::comparable<channel, actor> {
public:
/**
* @brief Enqueues @p msg to the list of received messages.
*/
virtual void enqueue(const message_header& hdr, any_tuple msg) = 0;
/**
* @brief Enqueues @p msg to the list of received messages and returns
* true if this is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_enqueue(const message_header& hdr, any_tuple msg);
/**
* @brief Sets the state from @p pending to @p ready.
*/
virtual void unchain();
protected:
virtual ~channel();
channel() = default;
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*);
explicit operator bool() const;
bool operator!() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
intptr_t compare(const channel& other) const;
intptr_t compare(const actor& other) const;
static intptr_t compare(const abstract_channel* lhs, const abstract_channel* rhs);
private:
intrusive_ptr<abstract_channel> m_ptr;
};
/**
* @brief A smart pointer type that manages instances of {@link channel}.
* @relates channel
*/
typedef intrusive_ptr<channel> channel_ptr;
/**
* @brief Convenience alias.
*/
template<typename T, typename R = void>
using enable_if_channel = std::enable_if<std::is_base_of<channel, T>::value, R>;
} // namespace cppa
#endif // CPPA_CHANNEL_HPP
......@@ -62,7 +62,7 @@ class context_switching_actor : public extend<scheduled_actor, context_switching
*/
context_switching_actor(std::function<void()> fun);
resume_result resume(util::fiber* from, actor_ptr& next_job);
resume_result resume(util::fiber* from);
scheduled_actor_type impl_type();
......
......@@ -37,7 +37,6 @@
#include <type_traits>
#include "cppa/get.hpp"
#include "cppa/actor.hpp"
#include "cppa/cow_ptr.hpp"
#include "cppa/ref_counted.hpp"
......
......@@ -40,14 +40,9 @@
#include "cppa/on.hpp"
#include "cppa/atom.hpp"
#include "cppa/send.hpp"
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/match.hpp"
#include "cppa/spawn.hpp"
#include "cppa/channel.hpp"
#include "cppa/receive.hpp"
#include "cppa/factory.hpp"
#include "cppa/behavior.hpp"
#include "cppa/announce.hpp"
#include "cppa/sb_actor.hpp"
......@@ -63,9 +58,9 @@
#include "cppa/local_actor.hpp"
#include "cppa/prioritizing.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/typed_actor_ptr.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -84,7 +79,6 @@
#include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/receive_loop_helper.hpp"
/**
* @author Dominik Charousset <dominik.charousset (at) haw-hamburg.de>
......@@ -440,47 +434,32 @@
namespace cppa {
/**
* @ingroup MessageHandling
* @{
*/
/**
* @brief Sends a message to @p whom.
*
* <b>Usage example:</b>
* @code
* self << make_any_tuple(1, 2, 3);
* @endcode
* @param whom Receiver of the message.
* @param what Message as instance of {@link any_tuple}.
* @returns @p whom.
*/
template<class C>
inline typename enable_if_channel<C, const intrusive_ptr<C>&>::type
operator<<(const intrusive_ptr<C>& whom, any_tuple what) {
send_tuple(whom, std::move(what));
return whom;
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));
}
inline const self_type& operator<<(const self_type& s, any_tuple what) {
send_tuple(s.get(), std::move(what));
return s;
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) {
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) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
}
/**
* @brief Blocks execution of this actor until all
* other actors finished execution.
* @warning This function will cause a deadlock if called from multiple actors.
* @warning Do not call this function in cooperatively scheduled actors.
*/
inline void await_all_others_done() {
auto value = (self.unchecked() == nullptr) ? 0 : 1;
get_actor_registry()->await_running_count_equal(value);
inline void await_all_actors_done() {
get_actor_registry()->await_running_count_equal(0);
}
/**
......@@ -493,7 +472,7 @@ inline void await_all_others_done() {
* @p nullptr.
* @throws bind_failure
*/
void publish(actor_ptr whom, std::uint16_t port, const char* addr = nullptr);
void publish(actor whom, std::uint16_t port, const char* addr = nullptr);
/**
* @brief Publishes @p whom using @p acceptor to handle incoming connections.
......@@ -502,7 +481,7 @@ void publish(actor_ptr whom, std::uint16_t port, const char* addr = nullptr);
* @param whom Actor that should be published at @p port.
* @param acceptor Network technology-specific acceptor implementation.
*/
void publish(actor_ptr whom, std::unique_ptr<io::acceptor> acceptor);
void publish(actor whom, std::unique_ptr<io::acceptor> acceptor);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
......@@ -511,12 +490,12 @@ void publish(actor_ptr whom, std::unique_ptr<io::acceptor> acceptor);
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
*/
actor_ptr remote_actor(const char* host, std::uint16_t port);
actor remote_actor(const char* host, std::uint16_t port);
/**
* @copydoc remote_actor(const char*, std::uint16_t)
*/
inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) {
inline actor remote_actor(const std::string& host, std::uint16_t port) {
return remote_actor(host.c_str(), port);
}
......@@ -527,7 +506,7 @@ inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) {
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
*/
actor_ptr remote_actor(io::stream_ptr_pair connection);
actor remote_actor(io::stream_ptr_pair connection);
/**
* @brief Spawns an IO actor of type @p Impl.
......@@ -537,9 +516,9 @@ actor_ptr 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_ptr spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
actor spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
using namespace io;
using namespace std;
auto ptr = make_counted<Impl>(move(in), move(out), forward<Ts>(args)...);
......@@ -555,10 +534,10 @@ actor_ptr spawn_io(io::input_stream_ptr in,
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor_ptr spawn_io(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
actor spawn_io(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
using namespace std;
auto ptr = io::broker::from(move(fun), move(in), move(out),
forward<Ts>(args)...);
......@@ -576,7 +555,7 @@ 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*)>,
typename... Ts>
actor_ptr spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
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, std::forward<Ts>(args)...);
}
......@@ -584,7 +563,7 @@ actor_ptr spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args)
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor_ptr spawn_io_server(F fun, uint16_t port, Ts&&... args) {
actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
using namespace std;
auto ptr = io::broker::from(move(fun), io::ipv4_acceptor::create(port),
forward<Ts>(args)...);
......@@ -599,33 +578,6 @@ actor_ptr spawn_io_server(F fun, uint16_t port, Ts&&... args) {
*/
void shutdown(); // note: implemented in singleton_manager.cpp
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
template<typename T, typename... Ts>
inline typename std::enable_if<
!is_behavior_policy<typename util::rm_const_and_ref<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
self->do_become(match_expr_convert(std::forward<T>(arg),
std::forward<Ts>(args)...),
true);
}
template<bool Discard, typename... Ts>
inline void become(behavior_policy<Discard>, Ts&&... args) {
self->do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
/**
* @brief Returns to a previous behavior if available.
*/
inline void unbecome() {
self->do_unbecome();
}
struct actor_ostream {
typedef const actor_ostream& (*fun_type)(const actor_ostream&);
......@@ -633,12 +585,12 @@ struct actor_ostream {
constexpr actor_ostream() { }
inline const actor_ostream& write(std::string arg) const {
send(get_scheduler()->printer(), atom("add"), move(arg));
send_as(nullptr, get_scheduler()->printer(), atom("add"), move(arg));
return *this;
}
inline const actor_ostream& flush() const {
send(get_scheduler()->printer(), atom("flush"));
send_as(nullptr, get_scheduler()->printer(), atom("flush"));
return *this;
}
......@@ -678,9 +630,9 @@ inline const actor_ostream& operator<<(const actor_ostream& o, actor_ostream::fu
namespace std {
// allow actor_ptr to be used in hash maps
template<>
struct hash<cppa::actor_ptr> {
inline size_t operator()(const cppa::actor_ptr& ptr) const {
return (ptr) ? static_cast<size_t>(ptr->id()) : 0;
struct hash<cppa::actor> {
inline size_t operator()(const cppa::actor& ref) const {
return static_cast<size_t>(ref.id());
}
};
// provide convenience overlaods for aout; implemented in logging.cpp
......
......@@ -39,15 +39,16 @@ namespace cppa {
class actor;
class group;
class channel;
class node_id;
class behavior;
class any_tuple;
class self_type;
class actor_proxy;
class abstract_actor;
class message_header;
class partial_function;
class uniform_type_info;
class primitive_variant;
class node_id;
// structs
struct anything;
......@@ -62,11 +63,9 @@ template<typename> class intrusive_ptr;
template<typename> class weak_intrusive_ptr;
// intrusive pointer typedefs
typedef intrusive_ptr<actor> actor_ptr;
typedef intrusive_ptr<group> group_ptr;
typedef intrusive_ptr<channel> channel_ptr;
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
typedef intrusive_ptr<node_id> node_id_ptr;
typedef intrusive_ptr<node_id> node_id_ptr;
// weak intrusive pointer typedefs
typedef weak_intrusive_ptr<actor_proxy> weak_actor_proxy_ptr;
......
......@@ -38,8 +38,8 @@
#include <cstdint>
#include <condition_variable>
#include "cppa/actor.hpp"
#include "cppa/attachable.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/detail/singleton_mixin.hpp"
......@@ -59,7 +59,7 @@ class actor_registry : public singleton_mixin<actor_registry> {
* exit reason. An entry with a nullptr means the actor has finished
* execution for given reason.
*/
typedef std::pair<actor_ptr, std::uint32_t> value_type;
typedef std::pair<abstract_actor_ptr, std::uint32_t> value_type;
/**
* @brief Returns the {nullptr, invalid_exit_reason}.
......@@ -67,11 +67,11 @@ class actor_registry : public singleton_mixin<actor_registry> {
value_type get_entry(actor_id key) const;
// return nullptr if the actor wasn't put *or* finished execution
inline actor_ptr get(actor_id key) const {
inline abstract_actor_ptr get(actor_id key) const {
return get_entry(key).first;
}
void put(actor_id key, const actor_ptr& value);
void put(actor_id key, const abstract_actor_ptr& value);
void erase(actor_id key, std::uint32_t reason);
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_EVENT_BASED_ACTOR_FACTORY_HPP
#define CPPA_EVENT_BASED_ACTOR_FACTORY_HPP
#include <type_traits>
#include "cppa/scheduler.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/detail/tdata.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/util/type_list.hpp"
namespace cppa { namespace detail {
template<typename InitFun, typename CleanupFun, typename... Members>
class event_based_actor_impl : public event_based_actor {
public:
template<typename... Ts>
event_based_actor_impl(InitFun fun, CleanupFun cfun, Ts&&... args)
: m_init(std::move(fun)), m_on_exit(std::move(cfun))
, m_members(std::forward<Ts>(args)...) { }
void init() { apply(m_init); }
void on_exit() override {
typedef typename util::get_callable_trait<CleanupFun>::arg_types arg_types;
std::integral_constant<size_t, util::tl_size<arg_types>::value> token;
on_exit_impl(m_on_exit, token);
}
private:
InitFun m_init;
CleanupFun m_on_exit;
tdata<Members...> m_members;
template<typename F>
void apply(F& f, typename std::add_pointer<Members>::type... args) {
f(args...);
}
template<typename F, typename... Ts>
void apply(F& f, Ts... args) {
apply(f, args..., &get_ref<sizeof...(Ts)>(m_members));
}
typedef std::integral_constant<size_t, 0> zero_t;
template<typename OnExit, typename Token>
typename std::enable_if<std::is_same<Token, zero_t>::value>::type
on_exit_impl(OnExit& fun, Token) {
fun();
}
template<typename OnExit, typename Token>
typename std::enable_if<std::is_same<Token, zero_t>::value == false>::type
on_exit_impl(OnExit& fun, Token) {
apply(fun);
}
};
template<typename InitFun, typename CleanupFun, typename... Members>
class event_based_actor_factory {
public:
typedef event_based_actor_impl<InitFun, CleanupFun, Members...> impl;
event_based_actor_factory(InitFun fun, CleanupFun cfun)
: m_init(std::move(fun)), m_on_exit(std::move(cfun)) { }
template<typename... Ts>
actor_ptr spawn(Ts&&... args) {
auto ptr = make_counted<impl>(m_init, m_on_exit, std::forward<Ts>(args)...);
return get_scheduler()->exec(no_spawn_options, ptr);
}
private:
InitFun m_init;
CleanupFun m_on_exit;
};
// event-based actor factory from type list
template<typename InitFun, typename CleanupFun, class TypeList>
struct ebaf_from_type_list;
template<typename InitFun, typename CleanupFun, typename... Ts>
struct ebaf_from_type_list<InitFun, CleanupFun, util::type_list<Ts...> > {
typedef event_based_actor_factory<InitFun, CleanupFun, Ts...> type;
};
template<typename Init, typename Cleanup>
struct ebaf_from_functor {
typedef typename util::get_callable_trait<Init>::arg_types arg_types;
typedef typename util::get_callable_trait<Cleanup>::arg_types arg_types2;
static_assert(util::tl_forall<arg_types, std::is_pointer>::value,
"First functor takes non-pointer arguments");
static_assert( std::is_same<arg_types, arg_types2>::value
|| util::tl_empty<arg_types2>::value,
"Second functor must provide either the same signature "
"as the first one or must take zero arguments");
typedef typename util::tl_map<arg_types, std::remove_pointer>::type mems;
typedef typename ebaf_from_type_list<Init, Cleanup, mems>::type type;
};
} } // namespace cppa::detail
#endif // CPPA_EVENT_BASED_ACTOR_FACTORY_HPP
......@@ -34,9 +34,6 @@
#include <string>
#include <type_traits>
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa { class local_actor; }
......@@ -76,7 +73,7 @@ struct implicit_conversions {
typedef typename util::replace_type<
subtype3,
actor_ptr,
actor,
std::is_convertible<T, actor*>,
std::is_convertible<T, local_actor*>,
std::is_same<self_type, T>
......
......@@ -379,7 +379,7 @@ class receive_policy {
<< " did not reply to a "
"synchronous request message");
auto fhdl = fetch_response_handle(client, hdl);
if (fhdl.valid()) fhdl.apply(atom("VOID"));
if (fhdl.valid()) fhdl.apply(make_any_tuple(atom("VOID")));
}
} else {
if ( matches<atom_value, std::uint64_t>(*res)
......
......@@ -38,7 +38,7 @@ namespace cppa { namespace detail {
struct scheduled_actor_dummy : scheduled_actor {
scheduled_actor_dummy();
void enqueue(const message_header&, any_tuple) override;
resume_result resume(util::fiber*, actor_ptr&) override;
resume_result resume(util::fiber*) override;
void quit(std::uint32_t) override;
void dequeue(behavior&) override;
void dequeue_response(behavior&, message_id) override;
......
......@@ -33,29 +33,22 @@
#include <cstdint>
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp"
namespace cppa {
class actor_addr;
class message_id;
class local_actor;
class mailbox_element;
} // namespace cppa
namespace cppa { namespace detail {
struct sync_request_bouncer {
std::uint32_t rsn;
constexpr sync_request_bouncer(std::uint32_t r)
: rsn(r == exit_reason::not_exited ? exit_reason::normal : r) { }
inline void operator()(const actor_ptr& sender, const message_id& mid) const {
CPPA_REQUIRE(rsn != exit_reason::not_exited);
if (mid.is_request() && sender != nullptr) {
sender->enqueue({nullptr, sender, mid.response_id()},
make_any_tuple(atom("EXITED"), rsn));
}
}
inline void operator()(const mailbox_element& e) const {
(*this)(e.sender, e.mid);
}
explicit sync_request_bouncer(std::uint32_t r);
void operator()(const actor_addr& sender, const message_id& mid) const;
void operator()(const mailbox_element& e) const;
};
} } // namespace cppa::detail
......
......@@ -47,7 +47,7 @@ class thread_pool_scheduler : public scheduler {
public:
using scheduler::init_callback;
using scheduler::void_function;
using scheduler::actor_fun;
struct worker;
......@@ -63,7 +63,7 @@ class thread_pool_scheduler : public scheduler {
local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) override;
local_actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f) override;
local_actor_ptr exec(spawn_options opts, init_callback init_cb, actor_fun f) override;
private:
......
......@@ -62,8 +62,8 @@ using mapped_type_list = util::type_list<
bool,
any_tuple,
atom_value,
actor_ptr,
channel_ptr,
actor,
channel,
group_ptr,
node_id_ptr,
io::accept_handle,
......
......@@ -57,17 +57,23 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> {
public:
resume_result resume(util::fiber*, actor_ptr&);
resume_result resume(util::fiber*);
/**
* @brief Initializes the actor.
*/
virtual void init() = 0;
virtual behavior make_behavior() = 0;
scheduled_actor_type impl_type();
static intrusive_ptr<event_based_actor> from(std::function<void()> fun);
static intrusive_ptr<event_based_actor> from(std::function<behavior()> fun);
static intrusive_ptr<event_based_actor> from(std::function<void(event_based_actor*)> fun);
static intrusive_ptr<event_based_actor> from(std::function<behavior(event_based_actor*)> fun);
protected:
event_based_actor(actor_state st = actor_state::blocked);
......
......@@ -37,6 +37,7 @@
#include "cppa/channel.hpp"
#include "cppa/attachable.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/abstract_channel.hpp"
namespace cppa { namespace detail {
......@@ -53,7 +54,7 @@ class deserializer;
/**
* @brief A multicast group.
*/
class group : public channel {
class group : public abstract_channel {
friend class detail::group_manager;
friend class detail::peer_connection; // needs access to remote_enqueue
......@@ -77,7 +78,7 @@ class group : public channel {
subscription() = default;
subscription(subscription&&) = default;
subscription(const channel_ptr& s, const intrusive_ptr<group>& g);
subscription(const channel& s, const intrusive_ptr<group>& g);
~subscription();
......@@ -85,7 +86,7 @@ class group : public channel {
private:
channel_ptr m_subscriber;
channel m_subscriber;
intrusive_ptr<group> m_group;
};
......@@ -146,7 +147,7 @@ class group : public channel {
* @returns A {@link subscription} object that unsubscribes @p who
* if the lifetime of @p who ends.
*/
virtual subscription subscribe(const channel_ptr& who) = 0;
virtual subscription subscribe(const channel& who) = 0;
/**
* @brief Get a pointer to the group associated with
......@@ -181,7 +182,7 @@ class group : public channel {
group(module_ptr module, std::string group_id);
virtual void unsubscribe(const channel_ptr& who) = 0;
virtual void unsubscribe(const channel& who) = 0;
module_ptr m_module;
std::string m_identifier;
......
......@@ -45,8 +45,6 @@
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp"
#include "cppa/detail/fwd.hpp"
namespace cppa { namespace io {
class broker;
......@@ -108,7 +106,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
return from_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
detail::fwd<Ts>(args)...),
std::forward<Ts>(args)...),
std::move(in),
std::move(out));
}
......@@ -119,19 +117,17 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
static broker_ptr from(F fun, acceptor_uptr in, T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...),
std::forward<T0>(arg0),
std::forward<Ts>(args)...),
std::move(in));
}
template<typename F, typename... Ts>
actor_ptr fork(F fun,
connection_handle hdl,
Ts&&... args) {
actor fork(F fun, connection_handle hdl, Ts&&... args) {
return this->fork_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
detail::fwd<Ts>(args)...),
std::forward<Ts>(args)...),
hdl);
}
......@@ -160,7 +156,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
private:
actor_ptr fork_impl(std::function<void (broker*)> fun,
actor_addr fork_impl(std::function<void (broker*)> fun,
connection_handle hdl);
static broker_ptr from_impl(std::function<void (broker*)> fun,
......
......@@ -137,7 +137,7 @@ class middleman {
virtual void deliver(const node_id& node,
const message_header& hdr,
any_tuple msg ) = 0;
/**
* @brief This callback is invoked by {@link peer} implementations
* and causes the middleman to disconnect from the node.
......@@ -156,7 +156,7 @@ class middleman {
* to the event loop of the middleman.
* @note This member function is thread-safe.
*/
virtual void register_acceptor(const actor_ptr& pa, peer_acceptor* ptr) = 0;
virtual void register_acceptor(const actor_addr& pa, peer_acceptor* ptr) = 0;
/**
* @brief Returns the namespace that contains all remote actors
......
......@@ -123,18 +123,18 @@ class peer : public extend<continuable>::with<buffered_writing> {
type_lookup_table m_incoming_types;
type_lookup_table m_outgoing_types;
void monitor(const actor_ptr& sender, const node_id_ptr& node, actor_id aid);
void monitor(const actor_addr& sender, const node_id_ptr& node, actor_id aid);
void kill_proxy(const actor_ptr& sender, const node_id_ptr& node, actor_id aid, std::uint32_t reason);
void kill_proxy(const actor_addr& sender, const node_id_ptr& node, actor_id aid, std::uint32_t reason);
void link(const actor_ptr& sender, const actor_ptr& ptr);
void link(const actor_addr& sender, const actor_addr& ptr);
void unlink(const actor_ptr& sender, const actor_ptr& ptr);
void unlink(const actor_addr& sender, const actor_addr& ptr);
void deliver(const message_header& hdr, any_tuple msg);
inline void enqueue(const any_tuple& msg) {
enqueue({nullptr, nullptr}, msg);
enqueue({invalid_actor_addr, nullptr}, msg);
}
void enqueue_impl(const message_header& hdr, const any_tuple& msg);
......
......@@ -49,10 +49,10 @@ class peer_acceptor : public continuable {
continue_reading_result continue_reading() override;
peer_acceptor(middleman* parent,
acceptor_uptr ptr,
const actor_ptr& published_actor);
acceptor_uptr ptr,
const actor_addr& published_actor);
inline const actor_ptr& published_actor() const { return m_pa; }
inline const actor_addr& published_actor() const { return m_pa; }
void dispose() override;
......@@ -62,7 +62,7 @@ class peer_acceptor : public continuable {
middleman* m_parent;
acceptor_uptr m_ptr;
actor_ptr m_pa;
actor_addr m_pa;
};
......
......@@ -57,13 +57,13 @@ class sync_request_info : public extend<memory_managed>::with<memory_cached> {
typedef sync_request_info* pointer;
pointer next; // intrusive next pointer
actor_ptr sender; // points to the sender of the message
pointer next; // intrusive next pointer
actor_addr sender; // points to the sender of the message
message_id mid; // sync message ID
private:
sync_request_info(actor_ptr sptr, message_id id);
sync_request_info(actor_addr sptr, message_id id);
};
......@@ -79,17 +79,17 @@ class remote_actor_proxy : public actor_proxy {
void enqueue(const message_header& hdr, any_tuple msg) override;
void link_to(const actor_ptr& other) override;
void link_to(const actor_addr& other) override;
void unlink_from(const actor_ptr& other) override;
void unlink_from(const actor_addr& other) override;
bool remove_backlink(const actor_ptr& to) override;
bool remove_backlink(const actor_addr& to) override;
bool establish_backlink(const actor_ptr& to) override;
bool establish_backlink(const actor_addr& to) override;
void local_link_to(const actor_ptr& other) override;
void local_link_to(const actor_addr& other) override;
void local_unlink_from(const actor_ptr& other) override;
void local_unlink_from(const actor_addr& other) override;
void deliver(const message_header& hdr, any_tuple msg) override;
......
This diff is collapsed.
......@@ -36,7 +36,6 @@
#include <execinfo.h>
#include <type_traits>
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/singletons.hpp"
#include "cppa/local_actor.hpp"
......@@ -70,7 +69,7 @@ class logging {
const char* function_name,
const char* file_name,
int line_num,
const actor_ptr& from,
actor_addr from,
const std::string& msg ) = 0;
class trace_helper {
......@@ -81,7 +80,7 @@ class logging {
const char* fun_name,
const char* file_name,
int line_num,
actor_ptr aptr,
actor_addr aptr,
const std::string& msg);
~trace_helper();
......@@ -92,7 +91,7 @@ class logging {
const char* m_fun_name;
const char* m_file_name;
int m_line_num;
actor_ptr m_self;
actor_addr m_self;
};
......@@ -110,14 +109,6 @@ class logging {
};
inline actor_ptr fwd_aptr(const self_type& s) {
return s.unchecked();
}
inline actor_ptr fwd_aptr(actor_ptr ptr) {
return ptr;
}
struct oss_wr {
inline oss_wr() { }
......
......@@ -33,9 +33,9 @@
#include <cstdint>
#include "cppa/actor.hpp"
#include "cppa/extend.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/memory_cached.hpp"
......@@ -57,7 +57,7 @@ class mailbox_element : public extend<memory_managed>::with<memory_cached> {
pointer next; // intrusive next pointer
bool marked; // denotes if this node is currently processed
actor_ptr sender; // points to the sender of msg
actor_addr sender;
any_tuple msg; // 'content field'
message_id mid;
......
......@@ -34,19 +34,13 @@
#include <cstdint>
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/atom.hpp"
#include "cppa/logging.hpp"
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/util/type_traits.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/partial_function.hpp"
namespace cppa {
class local_actor;
namespace detail { struct optional_any_tuple_visitor; }
/**
......@@ -62,24 +56,8 @@ class continue_helper {
inline continue_helper(message_id mid) : m_mid(mid) { }
template<typename F>
continue_helper& continue_with(F fun) {
return continue_with(behavior::continuation_fun{partial_function{
on(any_vals, arg_match) >> fun
}});
}
inline continue_helper& continue_with(behavior::continuation_fun fun) {
auto ref_opt = self->bhvr_stack().sync_handler(m_mid);
if (ref_opt) {
auto& ref = *ref_opt;
// copy original behavior
behavior cpy = ref;
ref = cpy.add_continuation(std::move(fun));
}
else CPPA_LOG_ERROR(".continue_with: failed to add continuation");
return *this;
}
continue_helper& continue_with(F fun);
inline message_id get_message_id() const {
return m_mid;
}
......@@ -87,6 +65,7 @@ class continue_helper {
private:
message_id m_mid;
local_actor* self;
};
......@@ -99,14 +78,16 @@ class message_future {
message_future() = delete;
continue_helper then(const partial_function& pfun);
continue_helper await(const partial_function& pfun);
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template<typename... Cs, typename... Ts>
continue_helper then(const match_expr<Cs...>& arg, const Ts&... args) {
check_consistency();
self->become_waiting_for(match_expr_convert(arg, args...), m_mid);
return {m_mid};
return then(match_expr_convert(arg, args...));
}
/**
......@@ -114,8 +95,7 @@ class message_future {
*/
template<typename... Cs, typename... Ts>
void await(const match_expr<Cs...>& arg, const Ts&... args) {
check_consistency();
self->dequeue_response(match_expr_convert(arg, args...), m_mid);
await(match_expr_convert(arg, args...));
}
/**
......@@ -129,9 +109,7 @@ class message_future {
continue_helper
>::type
then(Fs... fs) {
check_consistency();
self->become_waiting_for(fs2bhvr(std::move(fs)...), m_mid);
return {m_mid};
return then(fs2bhvr(std::move(fs)...));
}
/**
......@@ -142,8 +120,7 @@ class message_future {
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
check_consistency();
self->dequeue_response(fs2bhvr(std::move(fs)...), m_mid);
await(fs2bhvr({std::move(fs)...}));
}
/**
......@@ -159,117 +136,11 @@ class message_future {
private:
message_id m_mid;
local_actor* self;
template<typename... Fs>
behavior fs2bhvr(Fs... fs) {
auto handle_sync_timeout = []() -> match_hint {
self->handle_sync_timeout();
return match_hint::skip;
};
return {
on(atom("EXITED"), val<std::uint32_t>) >> skip_message,
on(atom("VOID")) >> skip_message,
on(atom("TIMEOUT")) >> handle_sync_timeout,
(on(any_vals, arg_match) >> fs)...
};
}
void check_consistency() {
if (!m_mid.valid() || !m_mid.is_response()) {
throw std::logic_error("handle does not point to a response");
}
else if (!self->awaits(m_mid)) {
throw std::logic_error("response already received");
}
}
};
template<typename R>
class typed_continue_helper {
public:
typedef int message_id_wrapper_tag;
typedef typename detail::lifted_result_type<R>::type result_types;
typed_continue_helper(continue_helper ch) : m_ch(std::move(ch)) { }
template<typename F>
typed_continue_helper<typename util::get_callable_trait<F>::result_type>
continue_with(F fun) {
detail::assert_types<result_types, F>();
m_ch.continue_with(std::move(fun));
return {m_ch};
}
inline message_id get_message_id() const {
return m_ch.get_message_id();
}
private:
continue_helper m_ch;
};
/*
template<>
class typed_continue_helper<void> {
public:
typedef int message_id_wrapper_tag;
typedef util::type_list<void> result_types;
typed_continue_helper(continue_helper ch) : m_ch(std::move(ch)) { }
template<typename F>
void continue_with(F fun) {
using arg_types = typename util::get_callable_trait<F>::arg_types;
static_assert(std::is_same<util::empty_type_list, arg_types>::value,
"functor takes too much arguments");
m_ch.continue_with(std::move(fun));
}
inline message_id get_message_id() const {
return m_ch.get_message_id();
}
private:
continue_helper m_ch;
};
*/
template<typename OutputList>
class typed_message_future {
public:
typed_message_future(message_future&& mf) : m_mf(std::move(mf)) { }
template<typename F>
void await(F fun) {
detail::assert_types<OutputList, F>();
m_mf.await(fun);
}
template<typename F>
typed_continue_helper<
typename util::get_callable_trait<F>::result_type
>
then(F fun) {
detail::assert_types<OutputList, F>();
return m_mf.then(fun);
}
private:
message_future m_mf;
partial_function fs2bhvr(partial_function pf);
void check_consistency();
};
......
......@@ -31,8 +31,8 @@
#ifndef CPPA_MESSAGE_HEADER_HPP
#define CPPA_MESSAGE_HEADER_HPP
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/message_priority.hpp"
......@@ -48,9 +48,9 @@ class message_header {
public:
actor_ptr sender;
channel_ptr receiver;
message_id id;
actor_addr sender;
channel receiver;
message_id id;
message_priority priority;
/**
......@@ -58,52 +58,12 @@ class message_header {
**/
message_header() = default;
/**
* @brief Creates a message header with <tt>receiver = dest</tt>
* and <tt>sender = self</tt>.
**/
template<typename T>
message_header(intrusive_ptr<T> dest)
: sender(self), receiver(dest), priority(message_priority::normal) {
static_assert(std::is_convertible<T*, channel*>::value,
"illegal receiver");
}
template<typename T>
message_header(T* dest)
: sender(self), receiver(dest), priority(message_priority::normal) {
static_assert(std::is_convertible<T*, channel*>::value,
"illegal receiver");
}
message_header(const std::nullptr_t&);
/**
* @brief Creates a message header with <tt>receiver = self</tt>
* and <tt>sender = self</tt>.
**/
message_header(const self_type&);
/**
* @brief Creates a message header with <tt>receiver = dest</tt>
* and <tt>sender = self</tt>.
*/
message_header(channel_ptr dest,
message_id mid,
message_priority prio = message_priority::normal);
/**
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = self</tt>.
*/
message_header(channel_ptr dest, message_priority prio);
/**
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = source</tt>.
*/
message_header(actor_ptr source,
channel_ptr dest,
message_header(actor_addr source,
channel dest,
message_id mid = message_id::invalid,
message_priority prio = message_priority::normal);
......@@ -111,7 +71,7 @@ class message_header {
* @brief Creates a message header with <tt>receiver = dest</tt> and
* <tt>sender = self</tt>.
*/
message_header(actor_ptr source, channel_ptr dest, message_priority prio);
message_header(actor_addr source, channel dest, message_priority prio);
void deliver(any_tuple msg) const;
......
......@@ -37,9 +37,9 @@
#include <algorithm>
#include <functional>
#include "cppa/actor.hpp"
#include "cppa/logging.hpp"
#include "cppa/opencl/global.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/opencl/smart_ptr.hpp"
#include "cppa/util/scope_guard.hpp"
......
......@@ -31,7 +31,7 @@
#ifndef OPTIONAL_VARIANT_HPP
#define OPTIONAL_VARIANT_HPP
#include <iosfwd>
#include <ostream>
#include <stdexcept>
#include <type_traits>
......
......@@ -50,8 +50,8 @@ class response_handle {
response_handle& operator=(response_handle&&) = default;
response_handle& operator=(const response_handle&) = default;
response_handle(const actor_ptr& from,
const actor_ptr& to,
response_handle(const actor_addr& from,
const actor_addr& to,
const message_id& response_id);
/**
......@@ -78,17 +78,17 @@ class response_handle {
/**
* @brief Returns the actor that is going send the response message.
*/
inline const actor_ptr& sender() const { return m_from; }
inline const actor_addr& sender() const { return m_from; }
/**
* @brief Returns the actor that is waiting for the response message.
*/
inline const actor_ptr& receiver() const { return m_to; }
inline const actor_addr& receiver() const { return m_to; }
private:
actor_ptr m_from;
actor_ptr m_to;
actor_addr m_from;
actor_addr m_to;
message_id m_id;
};
......
......@@ -88,7 +88,7 @@ class scheduled_actor : public extend<local_actor>::with<mailbox_based, threadle
* set in case of chaining.
* @note This member function is called from the scheduler's worker threads.
*/
virtual resume_result resume(util::fiber* from, actor_ptr& next_job) = 0;
virtual resume_result resume(util::fiber* from) = 0;
/**
* @brief Called once by the scheduler after actor is initialized,
......@@ -110,7 +110,7 @@ class scheduled_actor : public extend<local_actor>::with<mailbox_based, threadle
/**
* @brief Returns @p true if this actor is ignored by
* {@link await_all_others_done()}, false otherwise.
* {@link await_all_actors_done()}, false otherwise.
*/
inline bool is_hidden() const;
......@@ -118,10 +118,6 @@ class scheduled_actor : public extend<local_actor>::with<mailbox_based, threadle
void enqueue(const message_header&, any_tuple) override;
bool chained_enqueue(const message_header&, any_tuple) override;
void unchain() override;
protected:
scheduled_actor(actor_state init_state, bool enable_chained_send);
......
......@@ -44,17 +44,15 @@
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/spawn_options.hpp"
//#include "cppa/scheduled_actor.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/detail/fwd.hpp"
namespace cppa {
class self_type;
class scheduled_actor;
class scheduler_helper;
class event_based_actor;
typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr;
namespace detail { class singleton_manager; } // namespace detail
......@@ -86,9 +84,9 @@ class scheduler {
public:
typedef std::function<void(local_actor*)> init_callback;
typedef std::function<void()> void_function;
typedef std::function<behavior(local_actor*)> actor_fun;
const actor_ptr& printer() const;
const actor& printer() const;
virtual void enqueue(scheduled_actor*) = 0;
......@@ -115,7 +113,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
delayed_send_helper()->enqueue(nullptr, std::move(tup));
//TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
template<typename Duration, typename... Data>
......@@ -127,7 +125,7 @@ class scheduler {
util::duration{rel_time},
std::move(hdr),
std::move(data));
delayed_send_helper()->enqueue(nullptr, std::move(tup));
//TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
/**
......@@ -141,14 +139,13 @@ class scheduler {
* in this scheduler.
*/
virtual local_actor_ptr exec(spawn_options opts,
init_callback init_cb,
void_function actor_behavior) = 0;
init_callback init_cb,
actor_fun actor_behavior) = 0;
template<typename F, typename T, typename... Ts>
local_actor_ptr exec(spawn_options opts, init_callback cb,
F f, T&& a0, Ts&&... as) {
return this->exec(opts, cb, std::bind(f, detail::fwd<T>(a0),
detail::fwd<Ts>(as)...));
return this->exec(opts, cb, std::bind(f, std::forward<T>(a0), std::forward<Ts>(as)...));
}
private:
......@@ -157,7 +154,7 @@ class scheduler {
inline void dispose() { delete this; }
const actor_ptr& delayed_send_helper();
actor& delayed_send_helper();
scheduler_helper* m_helper;
......
......@@ -25,144 +25,131 @@
* *
* 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_RECEIVE_HPP
#define CPPA_RECEIVE_HPP
#ifndef CPPA_SCOPED_ACTOR_HPP
#define CPPA_SCOPED_ACTOR_HPP
#include "cppa/self.hpp"
#include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/detail/receive_loop_helper.hpp"
#include "cppa/thread_mapped_actor.hpp"
namespace cppa {
/**
* @ingroup BlockingAPI
* @{
*/
/**
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
template<typename... Ts>
void receive(Ts&&... args);
/**
* @brief Receives messages in an endless loop.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
template<typename... Ts>
void receive_loop(Ts&&... args);
/**
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename T>
detail::receive_for_helper<T> receive_for(T& begin, const T& end);
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
detail::receive_while_helper<Statement> receive_while(Statement&& stmt);
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
detail::do_receive_helper do_receive(Ts&&... args);
/**
* @}
*/
class scoped_actor {
public:
/**
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
template<typename... Ts>
void receive(Ts&&... args);
/**
* @brief Receives messages in an endless loop.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
template<typename... Ts>
void receive_loop(Ts&&... args);
/**
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename T>
detail::receive_for_helper<T> receive_for(T& begin, const T& end);
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
detail::receive_while_helper<Statement> receive_while(Statement&& stmt);
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
detail::do_receive_helper do_receive(Ts&&... args);
private:
intrusive_ptr<thread_mapped_actor> m_self;
};
/******************************************************************************
* inline and template function implementations *
******************************************************************************/
template<typename... Ts>
void receive(Ts&&... args) {
static_assert(sizeof...(Ts), "receive() requires at least one argument");
self->dequeue(match_expr_convert(std::forward<Ts>(args)...));
}
inline void receive_loop(behavior rules) {
local_actor* sptr = self;
for (;;) sptr->dequeue(rules);
void scoped_actor::receive(Ts&&... args) {
m_self->receive(std::forward<Ts(args)...);
}
template<typename... Ts>
void receive_loop(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
for (;;) sptr->dequeue(bhvr);
void scoped_actor::receive_loop(Ts&&... args) {
m_self->receive_loop(std::forward<Ts(args)...);
}
template<typename T>
detail::receive_for_helper<T> receive_for(T& begin, const T& end) {
return {begin, end};
detail::receive_for_helper<T> scoped_actor::receive_for(T& begin, const T& end) {
m_self->receive_for(begin, end);
}
template<typename Statement>
detail::receive_while_helper<Statement> receive_while(Statement&& stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean");
return std::move(stmt);
detail::receive_while_helper<Statement> scoped_actor::receive_while(Statement&& stmt) {
m_self->receive_while(std::forward<Statement(stmt));
}
template<typename... Ts>
detail::do_receive_helper do_receive(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
return detail::do_receive_helper(std::move(bhvr));
detail::do_receive_helper scoped_actor::do_receive(Ts&&... args) {
m_self->do_receive(std::forward<Ts(args)...);
}
} // namespace cppa
#endif // CPPA_RECEIVE_HPP
#endif // CPPA_SCOPED_ACTOR_HPP
......@@ -31,13 +31,11 @@
#ifndef CPPA_SEND_HPP
#define CPPA_SEND_HPP
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/message_header.hpp"
#include "cppa/message_future.hpp"
#include "cppa/typed_actor_ptr.hpp"
#include "cppa/util/duration.hpp"
......@@ -48,25 +46,9 @@ namespace cppa {
* @{
*/
/**
* @brief Stores sender, receiver, and message priority.
*/
template<typename T = channel_ptr>
struct destination_header {
T receiver;
message_priority priority;
template<typename U>
destination_header(U&& dest, message_priority mp = message_priority::normal)
: receiver(std::forward<U>(dest)), priority(mp) { }
destination_header(destination_header&&) = default;
destination_header(const destination_header&) = default;
destination_header& operator=(destination_header&&) = default;
destination_header& operator=(const destination_header&) = default;
};
using channel_destination = destination_header<channel_ptr>;
using actor_destination = destination_header<actor_ptr>;
using channel_destination = destination_header<channel>;
using actor_destination = destination_header<abstract_actor_ptr>;
/**
* @brief Sends @p what to the receiver specified in @p hdr.
......
......@@ -38,18 +38,15 @@
#include "cppa/typed_actor.hpp"
#include "cppa/prioritizing.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp"
namespace cppa {
/** @cond PRIVATE */
inline actor_ptr eval_sopts(spawn_options opts, local_actor_ptr ptr) {
CPPA_LOGF_INFO("spawned new local actor with ID " << ptr->id()
<< " of type " << detail::demangle(typeid(*ptr)));
if (has_monitor_flag(opts)) self->monitor(ptr);
if (has_link_flag(opts)) self->link_to(ptr);
return ptr;
constexpr bool unbound_spawn_options(spawn_options opts) {
return !has_monitor_flag(opts) && !has_link_flag(opts);
}
/** @endcond */
......@@ -63,15 +60,16 @@ inline actor_ptr eval_sopts(spawn_options opts, local_actor_ptr ptr) {
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn(Ts&&... args) {
actor spawn(Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
return eval_sopts(Options,
get_scheduler()->exec(Options,
scheduler::init_callback{},
std::forward<Ts>(args)...));
static_assert(unbound_spawn_options(Options),
"top-level spawns cannot have monitor or link flag");
return get_scheduler()->exec(Options,
scheduler::init_callback{},
std::forward<Ts>(args)...);
}
/**
......@@ -79,12 +77,14 @@ actor_ptr spawn(Ts&&... args) {
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @returns An {@link actor} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn(Ts&&... args) {
actor spawn(Ts&&... args) {
static_assert(std::is_base_of<event_based_actor, Impl>::value,
"Impl is not a derived type of event_based_actor");
static_assert(unbound_spawn_options(Options),
"top-level spawns cannot have monitor or link flag");
scheduled_actor_ptr ptr;
if (has_priority_aware_flag(Options)) {
using derived = typename extend<Impl>::template with<threaded, prioritizing>;
......@@ -95,7 +95,7 @@ actor_ptr spawn(Ts&&... args) {
ptr = make_counted<derived>(std::forward<Ts>(args)...);
}
else ptr = make_counted<Impl>(std::forward<Ts>(args)...);
return eval_sopts(Options, get_scheduler()->exec(Options, std::move(ptr)));
return get_scheduler()->exec(Options, std::move(ptr));
}
/**
......@@ -103,11 +103,11 @@ actor_ptr spawn(Ts&&... args) {
* immediately joins @p grp.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @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>
actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
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);
......@@ -123,11 +123,11 @@ actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @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_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
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));
......@@ -145,8 +145,9 @@ typename Impl::typed_pointer_type spawn_typed(Ts&&... args) {
);
}
/*TODO:
template<spawn_options Options, typename... Ts>
typed_actor_ptr<typename detail::deduce_signature<Ts>::type...>
typed_actor<typename detail::deduce_signature<Ts>::type...>
spawn_typed(const match_expr<Ts...>& me) {
static_assert(util::conjunction<
detail::match_expr_has_no_guard<Ts>::value...
......@@ -166,7 +167,7 @@ spawn_typed(const match_expr<Ts...>& me) {
}
template<typename... Ts>
typed_actor_ptr<typename detail::deduce_signature<Ts>::type...>
typed_actor<typename detail::deduce_signature<Ts>::type...>
spawn_typed(const match_expr<Ts...>& me) {
return spawn_typed<no_spawn_options>(me);
}
......@@ -180,6 +181,7 @@ auto spawn_typed(T0&& v0, T1&& v1, Ts&&... vs)
std::forward<T1>(v1),
std::forward<Ts>(vs)...));
}
*/
/** @} */
......
......@@ -93,7 +93,7 @@ constexpr spawn_options detached = spawn_options::detach_flag;
/**
* @brief Causes the runtime to ignore the new actor in
* {@link await_all_others_done()}.
* {@link await_all_actors_done()}.
*/
constexpr spawn_options hidden = spawn_options::hide_flag;
......
......@@ -41,7 +41,6 @@
#include "cppa/util/dptr.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/receive_policy.hpp"
namespace cppa {
......@@ -64,22 +63,140 @@ class stacked : public Base {
static constexpr auto receive_flag = detail::rp_nestable;
void dequeue(behavior& bhvr) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr);
--m_nested_count;
check_quit();
}
struct receive_while_helper {
void dequeue_response(behavior& bhvr, message_id request_id) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
--m_nested_count;
check_quit();
std::function<void(behavior&)> m_dq;
std::function<bool()> m_stmt;
template<typename... Ts>
void operator()(Ts&&... args) {
static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument");
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
while (m_stmt()) m_dq(bhvr);
}
};
template<typename T>
struct receive_for_helper {
std::function<void(behavior&)> m_dq;
T& begin;
T end;
template<typename... Ts>
void operator()(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for ( ; begin != end; ++begin) m_dq(bhvr);
}
};
struct do_receive_helper {
std::function<void(behavior&)> m_dq;
behavior m_bhvr;
template<typename Statement>
void until(Statement stmt) {
do { m_dq(m_bhvr); }
while (stmt() == false);
}
};
/**
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
template<typename... Ts>
void receive(Ts&&... args) {
static_assert(sizeof...(Ts), "receive() requires at least one argument");
dequeue(match_expr_convert(std::forward<Ts>(args)...));
}
/**
* @brief Receives messages in an endless loop.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for (;;) dequeue(bhvr);
}
/**
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename T>
receive_for_helper<T> receive_for(T& begin, const T& end) {
return {make_dequeue_callback(), begin, end};
}
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
receive_while_helper receive_while(Statement stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean");
return {make_dequeue_callback(), stmt};
}
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
do_receive_helper do_receive(Ts&&... args) {
return { make_dequeue_callback()
, match_expr_convert(std::forward<Ts>(args)...)};
}
virtual void run() {
exec_behavior_stack();
if (m_behavior) m_behavior();
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
......@@ -90,38 +207,13 @@ class stacked : public Base {
m_behavior = std::move(fun);
}
void exec_behavior_stack() override {
m_in_bhvr_loop = true;
auto dthis = util::dptr<Subtype>(this);
if (!stack_empty() && !m_nested_count) {
while (!stack_empty()) {
++m_nested_count;
dthis->m_bhvr_stack.exec(m_recv_policy, dthis);
--m_nested_count;
check_quit();
}
}
m_in_bhvr_loop = false;
}
protected:
template<typename... Ts>
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...)
, m_in_bhvr_loop(false)
, m_nested_count(0) { }
void do_become(behavior&& bhvr, bool discard_old) override {
become_impl(std::move(bhvr), discard_old, message_id());
}
void become_waiting_for(behavior bhvr, message_id mid) override {
become_impl(std::move(bhvr), false, mid);
}
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
virtual bool has_behavior() {
return static_cast<bool>(m_behavior)
|| !util::dptr<Subtype>(this)->m_bhvr_stack.empty();
return static_cast<bool>(m_behavior);
}
std::function<void()> m_behavior;
......@@ -129,43 +221,18 @@ class stacked : public Base {
private:
inline bool stack_empty() {
return util::dptr<Subtype>(this)->m_bhvr_stack.empty();
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
void become_impl(behavior&& bhvr, bool discard_old, message_id mid) {
auto dthis = util::dptr<Subtype>(this);
if (bhvr.timeout().valid()) {
dthis->reset_timeout();
dthis->request_timeout(bhvr.timeout());
}
if (!dthis->m_bhvr_stack.empty() && discard_old) {
dthis->m_bhvr_stack.pop_async_back();
}
dthis->m_bhvr_stack.push_back(std::move(bhvr), mid);
void dequeue(behavior& bhvr) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr);
}
void check_quit() {
// do nothing if other message handlers are still running
if (m_nested_count > 0) return;
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
if (rsn != exit_reason::not_exited) {
dthis->on_exit();
if (stack_empty()) {
dthis->cleanup(rsn);
throw actor_exited(rsn);
}
else {
dthis->planned_exit_reason(exit_reason::not_exited);
if (!m_in_bhvr_loop) exec_behavior_stack();
}
}
void dequeue_response(behavior& bhvr, message_id request_id) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
}
bool m_in_bhvr_loop;
size_t m_nested_count;
};
} // namespace cppa
......
......@@ -31,20 +31,52 @@
#ifndef CPPA_STACKLESS_HPP
#define CPPA_STACKLESS_HPP
#include "cppa/util/dptr.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @brief An actor that uses the non-blocking API of @p libcppa and
* does not has its own stack.
* @brief Policy tag that causes {@link event_based_actor::become} to
* discard the current behavior.
* @relates local_actor
*/
constexpr auto discard_behavior;
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* keep the current behavior available.
* @relates local_actor
*/
constexpr auto keep_behavior;
#else
template<bool DiscardBehavior>
struct behavior_policy { static constexpr bool discard_old = DiscardBehavior; };
template<typename T>
struct is_behavior_policy : std::false_type { };
template<bool DiscardBehavior>
struct is_behavior_policy<behavior_policy<DiscardBehavior>> : std::true_type { };
typedef behavior_policy<false> keep_behavior_t;
typedef behavior_policy<true > discard_behavior_t;
constexpr discard_behavior_t discard_behavior = discard_behavior_t{};
constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
#endif
template<class Base, class Subtype>
class stackless : public Base {
friend class detail::receive_policy;
friend class detail::behavior_stack;
protected:
typedef stackless combined_type;
......@@ -60,15 +92,30 @@ class stackless : public Base {
return this->m_bhvr_stack.empty() == false;
}
protected:
void do_become(behavior&& bhvr, bool discard_old) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
void unbecome() {
this->m_bhvr_stack.pop_async_back();
}
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
template<typename T, typename... Ts>
inline typename std::enable_if<
!is_behavior_policy<typename util::rm_const_and_ref<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
do_become(match_expr_convert(std::forward<T>(arg),
std::forward<Ts>(args)...),
true);
}
template<bool Discard, typename... Ts>
inline void become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
void become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
......@@ -76,16 +123,23 @@ class stackless : public Base {
}
this->m_bhvr_stack.push_back(std::move(bhvr), mf);
}
void do_become(behavior&& bhvr, bool discard_old) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
}
inline bool has_behavior() const {
return this->m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(this->m_bhvr_stack.empty() == false);
return this->m_bhvr_stack.back();
}
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
this->reset_timeout();
......@@ -95,44 +149,18 @@ class stackless : public Base {
}
}
// provoke compiler errors for usage of receive() and related functions
/**
* @brief Provokes a compiler error to ensure that a stackless actor
* does not accidently uses receive() instead of become().
*/
template<typename... Ts>
void receive(Ts&&...) {
// this asssertion always fails
static_assert((sizeof...(Ts) + 1) < 1,
"You shall not use receive in an event-based actor. "
"Use become() instead.");
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
receive(std::forward<Ts>(args)...);
void exec_bhvr_stack() {
while (!m_bhvr_stack.empty()) {
m_bhvr_stack.exec(m_recv_policy, util::dptr<Subtype>(this));
}
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_while(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
protected:
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void do_receive(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
// used for message handling in subclasses
detail::receive_policy m_recv_policy;
};
......
......@@ -35,6 +35,7 @@
#include <chrono>
#include <condition_variable>
#include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/util/dptr.hpp"
......@@ -93,10 +94,6 @@ class threaded : public Base {
// init() did indeed call quit() for some reason
dthis->on_exit();
}
while (!dthis->m_bhvr_stack.empty()) {
dthis->m_bhvr_stack.exec(dthis->m_recv_policy, dthis);
dthis->on_exit();
}
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
......@@ -137,11 +134,6 @@ class threaded : public Base {
enqueue_impl(this->m_mailbox, hdr, std::move(msg));
}
bool chained_enqueue(const message_header& hdr, any_tuple msg) override {
enqueue(hdr, std::move(msg));
return false;
}
timeout_type init_timeout(const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
......
......@@ -31,7 +31,9 @@
#ifndef CPPA_THREADLESS_HPP
#define CPPA_THREADLESS_HPP
#include "cppa/send.hpp"
#include "cppa/atom.hpp"
#include "cppa/behavior.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
......@@ -67,11 +69,11 @@ class threadless : public Base {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this, std::move(msg));
this->enqueue({this->address(), this}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else delayed_send_tuple(this, d, std::move(msg));
else this->delayed_send_tuple(this, d, std::move(msg));
m_has_pending_tout = true;
}
}
......
......@@ -68,7 +68,7 @@ inline std::string to_string(const message_header& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const actor_ptr& what) {
inline std::string to_string(const actor& what) {
return detail::to_string_impl(what);
}
......@@ -76,7 +76,7 @@ inline std::string to_string(const group_ptr& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const channel_ptr& what) {
inline std::string to_string(const channel& what) {
return detail::to_string_impl(what);
}
......
......@@ -25,75 +25,33 @@
* *
* 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_TYPED_ACTOR_HPP
#define CPPA_TYPED_ACTOR_HPP
#include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/message_future.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
namespace cppa {
template<typename... Signatures>
class typed_actor_ptr;
template<typename... Signatures>
class typed_actor : public event_based_actor {
public:
using signatures = util::type_list<Signatures...>;
using behavior_type = typed_behavior<Signatures...>;
class local_actor;
using typed_pointer_type = typed_actor_ptr<Signatures...>;
/**
* @brief Identifies an untyped actor.
*/
template<typename Interface>
class typed_actor {
protected:
virtual behavior_type make_behavior() = 0;
void init() final {
auto bhvr = make_behavior();
m_bhvr_stack.push_back(std::move(bhvr.unbox()));
}
void do_become(behavior&&, bool) final {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
}
};
} // namespace cppa
namespace cppa { namespace detail {
template<typename... Signatures>
class default_typed_actor : public typed_actor<Signatures...> {
public:
template<typename... Cases>
default_typed_actor(match_expr<Cases...> expr) : m_bhvr(std::move(expr)) { }
protected:
typed_behavior<Signatures...> make_behavior() override {
return m_bhvr;
}
friend class local_actor;
private:
typed_behavior<Signatures...> m_bhvr;
intrusive_ptr<abstract_actor> m_ptr;
};
} } // namespace cppa::detail
} // namespace cppa
#endif // CPPA_TYPED_ACTOR_HPP
......@@ -28,81 +28,72 @@
\******************************************************************************/
#ifndef CPPA_RECEIVE_LOOP_HELPER_HPP
#define CPPA_RECEIVE_LOOP_HELPER_HPP
#ifndef CPPA_TYPED_ACTOR_HPP
#define CPPA_TYPED_ACTOR_HPP
#include <new>
#include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/message_future.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/self.hpp"
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/util/type_list.hpp"
namespace cppa {
namespace cppa { namespace detail {
template<typename... Signatures>
class typed_actor_ptr;
template<typename Statement>
struct receive_while_helper {
Statement m_stmt;
template<typename... Signatures>
class typed_actor : public event_based_actor {
template<typename S>
receive_while_helper(S&& stmt) : m_stmt(std::forward<S>(stmt)) { }
public:
template<typename... Ts>
void operator()(Ts&&... args) {
static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument");
behavior tmp = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
while (m_stmt()) sptr->dequeue(tmp);
}
using signatures = util::type_list<Signatures...>;
};
using behavior_type = typed_behavior<Signatures...>;
template<typename T>
class receive_for_helper {
using typed_pointer_type = typed_actor_ptr<Signatures...>;
T& begin;
T end;
protected:
public:
virtual behavior_type make_behavior() = 0;
receive_for_helper(T& first, const T& last) : begin(first), end(last) { }
void init() final {
auto bhvr = make_behavior();
m_bhvr_stack.push_back(std::move(bhvr.unbox()));
}
template<typename... Ts>
void operator()(Ts&&... args) {
auto tmp = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
for ( ; begin != end; ++begin) sptr->dequeue(tmp);
void do_become(behavior&&, bool) final {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
}
};
class do_receive_helper {
} // namespace cppa
namespace cppa { namespace detail {
behavior m_bhvr;
template<typename... Signatures>
class default_typed_actor : public typed_actor<Signatures...> {
public:
inline do_receive_helper(behavior&& bhvr) : m_bhvr(std::move(bhvr)) { }
template<typename... Cases>
default_typed_actor(match_expr<Cases...> expr) : m_bhvr(std::move(expr)) { }
do_receive_helper(do_receive_helper&&) = default;
protected:
template<typename Statement>
void until(Statement&& stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean");
local_actor* sptr = self;
do {
sptr->dequeue(m_bhvr);
}
while (stmt() == false);
typed_behavior<Signatures...> make_behavior() override {
return m_bhvr;
}
private:
typed_behavior<Signatures...> m_bhvr;
};
} } // namespace cppa::detail
#endif // CPPA_RECEIVE_LOOP_HELPER_HPP
#endif // CPPA_TYPED_ACTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_UNTYPED_ACTOR_HPP
#define CPPA_UNTYPED_ACTOR_HPP
#include "cppa/extend.hpp"
#include "cppa/stackless.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
class untyped_actor : extend<local_actor>::with<stackless> {
};
} // namespace cppa
#endif // CPPA_UNTYPED_ACTOR_HPP
......@@ -164,9 +164,9 @@ struct is_builtin {
atom_value,
any_tuple,
message_header,
actor_ptr,
actor,
group_ptr,
channel_ptr,
channel,
node_id_ptr
>::value;
};
......
......@@ -401,7 +401,7 @@ int main() {
aout << color::cyan
<< "await CURL; this may take a while (press CTRL+C again to abort)"
<< color::reset_endl;
await_all_others_done();
await_all_actors_done();
// shutdown libcppa
shutdown();
// shutdown CURL
......
......@@ -37,7 +37,7 @@ int main() {
// create another actor that calls 'hello_world(mirror_actor)'
spawn(hello_world, mirror_actor);
// wait until all other actors we have spawned are done
await_all_others_done();
await_all_actors_done();
// run cleanup code before exiting main
shutdown();
}
......@@ -74,7 +74,7 @@ void tester(const actor_ptr& testee) {
int main() {
spawn(tester, spawn(calculator));
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......
......@@ -73,7 +73,7 @@ void dancing_kirby() {
int main() {
spawn(dancing_kirby);
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -184,7 +184,7 @@ int main(int, char**) {
spawn<philosopher>(names[i], chopsticks[i], chopsticks[(i+1)%5]);
}
// real philosophers are never done
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -216,7 +216,7 @@ int main(int argc, char** argv) {
if (host.empty()) host = "localhost";
client_repl(host, port);
}
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -143,7 +143,7 @@ int main(int argc, char** argv) {
);
// force actor to quit
send_exit(client_actor, exit_reason::user_shutdown);
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -118,7 +118,7 @@ int main(int, char**) {
// send t a foo_pair2
send(t, foo_pair2{3, 4});
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......
......@@ -58,7 +58,7 @@ int main(int, char**) {
make_pair(&foo::b, &foo::set_b));
auto t = spawn(testee);
send(t, foo{1, 2});
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -79,7 +79,7 @@ int main(int, char**) {
// spawn a new testee and send it a foo
send(spawn(testee), foo{1, 2});
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......
......@@ -129,7 +129,7 @@ int main(int, char**) {
auto t = spawn(testee, 2);
send(t, bar{foo{1, 2}, 3});
send(t, baz{foo{1, 2}, bar{foo{3, 4}, 5}});
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......
......@@ -201,7 +201,7 @@ int main() {
tvec.push_back(t0);
send(t, tvec);
await_all_others_done();
await_all_actors_done();
shutdown();
return 0;
}
......@@ -28,29 +28,10 @@
\******************************************************************************/
#ifndef FWD_HPP
#define FWD_HPP
#include "cppa/abstract_channel.hpp"
#include "cppa/self.hpp"
namespace cppa {
namespace cppa { namespace detail {
abstract_channel::~abstract_channel() { }
template<typename T>
struct is_self {
typedef typename util::rm_const_and_ref<T>::type plain_type;
static constexpr bool value = std::is_same<plain_type, self_type>::value;
};
template<typename T, typename U>
auto fwd(U& arg, typename std::enable_if<!is_self<T>::value>::type* = 0)
-> decltype(std::forward<T>(arg)) {
return std::forward<T>(arg);
}
template<typename T, typename U>
local_actor* fwd(U& arg, typename std::enable_if<is_self<T>::value>::type* = 0){
return arg;
}
} } // namespace cppa::detail
#endif // FWD_HPP
} // namespace cppa
......@@ -25,182 +25,38 @@
* *
* 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/config.hpp"
#include <utility>
#include <map>
#include <mutex>
#include <atomic>
#include <stdexcept>
#include "cppa/send.hpp"
#include "cppa/actor.hpp"
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/singletons.hpp"
#include "cppa/message_header.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
namespace { typedef std::unique_lock<std::mutex> guard_type; }
// m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
actor::actor(actor_id aid)
: m_id(aid), m_is_proxy(true), m_exit_reason(exit_reason::not_exited) { }
actor::actor()
: m_id(get_actor_registry()->next_id()), m_is_proxy(false)
, m_exit_reason(exit_reason::not_exited) { }
bool actor::link_to_impl(const actor_ptr& other) {
if (other && other != this) {
guard_type guard{m_mtx};
// send exit message if already exited
if (exited()) {
send_as(this, other, atom("EXIT"), exit_reason());
}
// add link if not already linked to other
// (checked by establish_backlink)
else if (other->establish_backlink(this)) {
m_links.push_back(other);
return true;
}
}
return false;
}
bool actor::attach(attachable_ptr ptr) {
if (ptr == nullptr) {
guard_type guard{m_mtx};
return m_exit_reason == exit_reason::not_exited;
}
std::uint32_t reason;
{ // lifetime scope of guard
guard_type guard{m_mtx};
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
m_attachables.push_back(std::move(ptr));
return true;
}
}
ptr->actor_exited(reason);
return false;
}
void actor::detach(const attachable::token& what) {
attachable_ptr ptr;
{ // lifetime scope of guard
guard_type guard{m_mtx};
auto end = m_attachables.end();
auto i = std::find_if(
m_attachables.begin(), end,
[&](attachable_ptr& p) { return p->matches(what); });
if (i != end) {
ptr = std::move(*i);
m_attachables.erase(i);
}
}
// ptr will be destroyed here, without locked mutex
}
void actor::link_to(const actor_ptr& other) {
static_cast<void>(link_to_impl(other));
}
actor::actor(abstract_actor* ptr) : m_ptr(ptr) { }
void actor::unlink_from(const actor_ptr& other) {
static_cast<void>(unlink_from_impl(other));
actor_id actor::id() const {
return (m_ptr) ? m_ptr->id() : 0;
}
bool actor::remove_backlink(const actor_ptr& other) {
if (other && other != this) {
guard_type guard{m_mtx};
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i != m_links.end()) {
m_links.erase(i);
return true;
}
}
return false;
actor_addr actor::address() const {
return m_ptr ? m_ptr->address() : actor_addr{};
}
bool actor::establish_backlink(const actor_ptr& other) {
std::uint32_t reason = exit_reason::not_exited;
if (other && other != this) {
guard_type guard{m_mtx};
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i == m_links.end()) {
m_links.push_back(other);
return true;
}
}
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
send_as(this, other, make_any_tuple(atom("EXIT"), reason));
}
return false;
void actor::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
bool actor::unlink_from_impl(const actor_ptr& other) {
guard_type guard{m_mtx};
// remove_backlink returns true if this actor is linked to other
if (other && !exited() && other->remove_backlink(this)) {
auto i = std::find(m_links.begin(), m_links.end(), other);
CPPA_REQUIRE(i != m_links.end());
m_links.erase(i);
return true;
}
return false;
bool actor::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
}
void actor::cleanup(std::uint32_t reason) {
// log as 'actor'
CPPA_LOGM_TRACE("cppa::actor", CPPA_ARG(m_id) << ", " << CPPA_ARG(reason));
CPPA_REQUIRE(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
decltype(m_links) mlinks;
decltype(m_attachables) mattachables;
{ // lifetime scope of guard
guard_type guard{m_mtx};
if (m_exit_reason != exit_reason::not_exited) {
// already exited
return;
}
m_exit_reason = reason;
mlinks = std::move(m_links);
mattachables = std::move(m_attachables);
// make sure lists are empty
m_links.clear();
m_attachables.clear();
}
CPPA_LOGC_INFO_IF(not is_proxy(), "cppa::actor", __func__,
"actor with ID " << m_id << " had " << mlinks.size()
<< " links and " << mattachables.size()
<< " attached functors; exit reason = " << reason
<< ", class = " << detail::demangle(typeid(*this)));
// send exit messages
auto msg = make_any_tuple(atom("EXIT"), reason);
CPPA_LOGM_DEBUG("cppa::actor", "send EXIT to " << mlinks.size() << " links");
for (actor_ptr& aptr : mlinks) {
message_header hdr{this, aptr, message_priority::high};
hdr.deliver(msg);
}
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
}
intptr_t actor::compare(const actor& other) const {
return channel::compare(m_ptr.get(), other.m_ptr.get());
}
} // namespace cppa
......@@ -25,128 +25,55 @@
* *
* 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 <pthread.h>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/actor.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
namespace cppa {
namespace {
pthread_key_t s_key;
pthread_once_t s_key_once = PTHREAD_ONCE_INIT;
local_actor* tss_constructor() {
local_actor* result = detail::memory::create<thread_mapped_actor>();
CPPA_LOGF_INFO("converted thread to actor; ID = " << result->id());
result->ref();
get_scheduler()->register_converted_context(result);
return result;
}
void tss_destructor(void* vptr) {
if (vptr) self_type::cleanup_fun(reinterpret_cast<local_actor*>(vptr));
}
void tss_make_key() {
pthread_key_create(&s_key, tss_destructor);
}
local_actor* tss_get() {
pthread_once(&s_key_once, tss_make_key);
return reinterpret_cast<local_actor*>(pthread_getspecific(s_key));
}
local_actor* tss_release() {
pthread_once(&s_key_once, tss_make_key);
auto result = reinterpret_cast<local_actor*>(pthread_getspecific(s_key));
if (result) {
pthread_setspecific(s_key, nullptr);
}
return result;
intptr_t compare_impl(const abstract_actor* lhs, const abstract_actor* rhs) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
local_actor* tss_get_or_create() {
pthread_once(&s_key_once, tss_make_key);
auto result = reinterpret_cast<local_actor*>(pthread_getspecific(s_key));
if (!result) {
result = tss_constructor();
pthread_setspecific(s_key, reinterpret_cast<void*>(result));
}
return result;
}
void tss_reset(local_actor* ptr, bool inc_ref_count = true) {
pthread_once(&s_key_once, tss_make_key);
auto old_ptr = reinterpret_cast<local_actor*>(pthread_getspecific(s_key));
if (old_ptr) {
tss_destructor(old_ptr);
}
if (ptr != nullptr && inc_ref_count) {
ptr->ref();
}
pthread_setspecific(s_key, reinterpret_cast<void*>(ptr));
}
} // namespace <anonymous>
void self_type::cleanup_fun(cppa::local_actor* what) {
if (what) {
auto ptr = dynamic_cast<thread_mapped_actor*>(what);
if (ptr) {
// make sure "unspawned" actors quit properly
what->cleanup(cppa::exit_reason::normal);
}
what->deref();
}
}
actor_addr::actor_addr(const invalid_actor_addr_t&) : m_ptr(nullptr) { }
void self_type::set_impl(self_type::pointer ptr) {
tss_reset(ptr);
}
actor_addr::actor_addr(abstract_actor* ptr) : m_ptr(ptr) { }
self_type::pointer self_type::get_impl() {
return tss_get_or_create();
actor_addr::operator bool() const {
return static_cast<bool>(m_ptr);
}
actor* self_type::convert_impl() {
return get_impl();
bool actor_addr::operator!() const {
return !m_ptr;
}
self_type::pointer self_type::get_unchecked_impl() {
return tss_get();
intptr_t actor_addr::compare(const actor& other) const {
return compare_impl(m_ptr.get(), other.m_ptr.get());
}
void self_type::adopt_impl(self_type::pointer ptr) {
tss_reset(ptr, false);
intptr_t actor_addr::compare(const actor_addr& other) const {
return compare_impl(m_ptr.get(), other.m_ptr.get());
}
self_type::pointer self_type::release_impl() {
return tss_release();
intptr_t actor_addr::compare(const local_actor* other) const {
return compare_impl(m_ptr.get(), other);
}
bool operator==(const actor_ptr& lhs, const self_type& rhs) {
return lhs.get() == rhs.get();
actor_id actor_addr::id() const {
return m_ptr->id();
}
bool operator==(const self_type& lhs, const actor_ptr& rhs) {
return rhs == lhs;
const node_id& actor_addr::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
}
bool operator!=(const actor_ptr& lhs, const self_type& rhs) {
return !(lhs == rhs);
bool actor_addr::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
}
bool operator!=(const self_type& lhs, const actor_ptr& rhs) {
return !(rhs == lhs);
}
} // namespace cppa
......@@ -44,32 +44,26 @@
namespace cppa {
void actor_namespace::write(serializer* sink, const actor_ptr& ptr) {
void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
CPPA_REQUIRE(sink != nullptr);
if (ptr == nullptr) {
if (!ptr) {
CPPA_LOG_DEBUG("serialize nullptr");
sink->write_value(static_cast<actor_id>(0));
node_id::serialize_invalid(sink);
}
else {
// local actor?
if (!ptr->is_proxy()) {
get_actor_registry()->put(ptr->id(), ptr);
// 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));
}
auto pinf = node_id::get();
if (ptr->is_proxy()) {
auto dptr = ptr.downcast<io::remote_actor_proxy>();
if (dptr) pinf = dptr->process_info();
else CPPA_LOG_ERROR("downcast failed");
}
sink->write_value(ptr->id());
sink->write_value(pinf->process_id());
sink->write_raw(node_id::host_id_size,
pinf->host_id().data());
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());
}
}
actor_ptr actor_namespace::read(deserializer* source) {
actor_addr actor_namespace::read(deserializer* source) {
CPPA_REQUIRE(source != nullptr);
node_id::host_id_type nid;
auto aid = source->read<uint32_t>();
......@@ -78,16 +72,15 @@ actor_ptr actor_namespace::read(deserializer* source) {
// local actor?
auto pinf = node_id::get();
if (aid == 0 && pid == 0) {
return nullptr;
return invalid_actor_addr;
}
else if (pid == pinf->process_id() && nid == pinf->host_id()) {
return get_actor_registry()->get(aid);
return actor{get_actor_registry()->get(aid)};
}
else {
node_id_ptr tmp = new node_id{pid, nid};
return get_or_put(tmp, aid);
return actor{get_or_put(tmp, aid)};
}
return nullptr;
}
size_t actor_namespace::count_proxies(const node_id& node) {
......@@ -95,7 +88,7 @@ size_t actor_namespace::count_proxies(const node_id& node) {
return (i != m_proxies.end()) ? i->second.size() : 0;
}
actor_ptr actor_namespace::get(const node_id& node, actor_id aid) {
actor_proxy_ptr actor_namespace::get(const node_id& node, actor_id aid) {
auto& submap = m_proxies[node];
auto i = submap.find(aid);
if (i != submap.end()) {
......@@ -109,7 +102,7 @@ actor_ptr actor_namespace::get(const node_id& node, actor_id aid) {
return nullptr;
}
actor_ptr actor_namespace::get_or_put(node_id_ptr node, actor_id aid) {
actor_proxy_ptr actor_namespace::get_or_put(node_id_ptr node, actor_id aid) {
auto result = get(*node, aid);
if (result == nullptr && m_factory) {
auto ptr = m_factory(aid, node);
......@@ -127,13 +120,6 @@ void actor_namespace::put(const node_id& node,
if (i == submap.end()) {
submap.insert(std::make_pair(aid, proxy));
if (m_new_element_callback) m_new_element_callback(aid, node);
/*if (m_parent) {
m_parent->enqueue(node,
{nullptr, nullptr},
make_any_tuple(atom("MONITOR"),
node_id::get(),
aid));
}*/
}
else {
CPPA_LOG_ERROR("proxy for " << aid << ":"
......@@ -151,7 +137,7 @@ void actor_namespace::erase(node_id& inf) {
}
void actor_namespace::erase(node_id& inf, actor_id aid) {
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
CPPA_LOG_TRACE(CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
auto i = m_proxies.find(inf);
if (i != m_proxies.end()) {
i->second.erase(aid);
......
......@@ -32,7 +32,6 @@
#include <limits>
#include <stdexcept>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/attachable.hpp"
#include "cppa/exit_reason.hpp"
......@@ -62,7 +61,7 @@ actor_registry::value_type actor_registry::get_entry(actor_id key) const {
return {nullptr, exit_reason::not_exited};
}
void actor_registry::put(actor_id key, const actor_ptr& value) {
void actor_registry::put(actor_id key, const abstract_actor_ptr& value) {
bool add_attachable = false;
if (value != nullptr) {
shared_guard guard(m_instances_mtx);
......
......@@ -38,7 +38,6 @@
#include <stdexcept>
#include <type_traits>
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/type_lookup_table.hpp"
#include "cppa/binary_deserializer.hpp"
......
......@@ -36,6 +36,7 @@
#include <cstring>
#include <type_traits>
#include "cppa/config.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/type_lookup_table.hpp"
......
......@@ -68,7 +68,7 @@ class default_broker : public broker {
: broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { }
void init() override {
enqueue(nullptr, make_any_tuple(atom("INITMSG")));
enqueue({invalid_actor_addr, this}, make_any_tuple(atom("INITMSG")));
become(
on(atom("INITMSG")) >> [=] {
unbecome();
......@@ -136,7 +136,7 @@ class broker::servant : public continuable {
if (!m_disconnected) {
m_disconnected = true;
if (m_broker->exit_reason() == exit_reason::not_exited) {
m_broker->invoke_message(nullptr, disconnect_message());
//TODO: m_broker->invoke_message(nullptr, disconnect_message());
}
}
}
......@@ -207,7 +207,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
|| m_policy == broker::exactly
|| m_policy == broker::at_most) {
CPPA_LOG_DEBUG("invoke io actor");
m_broker->invoke_message(nullptr, m_read_msg);
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_read_msg);
CPPA_LOG_INFO_IF(!m_read_msg.vals()->unique(), "detached buffer");
get_ref<2>(m_read_msg).clear();
}
......@@ -264,7 +264,7 @@ class broker::doorman : public broker::servant {
auto& p = *opt;
get_ref<2>(m_accept_msg) = m_broker->add_scribe(move(p.first),
move(p.second));
m_broker->invoke_message(nullptr, m_accept_msg);
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg);
}
else return read_continue_later;
}
......@@ -298,7 +298,6 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
m_dummy_node.mid = hdr.id;
try {
using detail::receive_policy;
scoped_self_setter sss{this};
auto bhvr = m_bhvr_stack.back();
switch (m_recv_policy.handle_message(this,
&m_dummy_node,
......@@ -335,7 +334,7 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
quit(exit_reason::unhandled_exception);
}
// restore dummy node
m_dummy_node.sender.reset();
m_dummy_node.sender = actor_addr{};
m_dummy_node.msg.reset();
}
......@@ -397,7 +396,6 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) {
}
local_actor_ptr init_and_launch(broker_ptr ptr) {
scoped_self_setter sss{ptr.get()};
ptr->init();
// continue reader only if not inherited from default_broker_impl
CPPA_LOGF_WARNING_IF(!ptr->has_behavior(), "broker w/o behavior spawned");
......@@ -454,8 +452,8 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) {
return id;
}
actor_ptr broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) {
actor_addr broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) {
auto i = m_io.find(hdl);
if (i == m_io.end()) throw std::invalid_argument("invalid handle");
scribe* sptr = i->second.get(); // non-owning pointer
......@@ -463,7 +461,7 @@ actor_ptr broker::fork_impl(std::function<void (broker*)> fun,
init_and_launch(result);
sptr->set_broker(result); // set new broker
m_io.erase(i);
return result;
return {result};
}
void broker::receive_policy(const connection_handle& hdl,
......
......@@ -28,18 +28,36 @@
\******************************************************************************/
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
namespace cppa {
channel::~channel() { }
intptr_t channel::compare(const abstract_channel* lhs, const abstract_channel* rhs) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
channel::channel(abstract_channel* ptr) : m_ptr(ptr) { }
bool channel::chained_enqueue(const message_header& hdr, any_tuple msg) {
enqueue(hdr, std::move(msg));
return false;
channel::operator bool() const {
return static_cast<bool>(m_ptr);
}
void channel::unchain() { }
bool channel::operator!() const {
return !m_ptr;
}
void channel::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
intptr_t channel::compare(const channel& other) const {
return compare(m_ptr.get(), other.m_ptr.get());
}
intptr_t channel::compare(const actor& other) const {
return compare(m_ptr.get(), other.m_ptr.get());
}
} // namespace cppa
} // namespace cppa
\ No newline at end of file
......@@ -32,7 +32,6 @@
#include "cppa/to_string.hpp"
#include "cppa/cppa.hpp"
#include "cppa/self.hpp"
#include "cppa/logging.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -46,34 +45,13 @@ class default_scheduled_actor : public event_based_actor {
public:
typedef std::function<void()> fun_type;
typedef std::function<behavior(event_based_actor*)> fun_type;
default_scheduled_actor(fun_type&& fun)
: super(actor_state::ready), m_fun(std::move(fun)), m_initialized(false) { }
void init() { }
resume_result resume(util::fiber* f, actor_ptr& next) {
if (!m_initialized) {
scoped_self_setter sss{this};
m_initialized = true;
m_fun();
if (m_bhvr_stack.empty()) {
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::normal);
}
else {
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
next.swap(m_chained_actor);
set_state(actor_state::done);
}
return resume_result::actor_done;
}
}
return event_based_actor::resume(f, next);
: super(actor_state::ready), m_fun(std::move(fun)) { }
behavior make_behavior() override {
return m_fun(this);
}
scheduled_actor_type impl_type() {
......@@ -83,21 +61,39 @@ class default_scheduled_actor : public event_based_actor {
private:
fun_type m_fun;
bool m_initialized;
};
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> fun) {
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<behavior()> fun) {
return make_counted<default_scheduled_actor>([=](event_based_actor*) -> behavior { return fun(); });
}
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<behavior(event_based_actor*)> fun) {
return make_counted<default_scheduled_actor>(std::move(fun));
}
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void(event_based_actor*)> fun) {
auto result = make_counted<default_scheduled_actor>([=](event_based_actor* self) -> behavior {
return (
on(atom("INIT")) >> [=] {
fun(self);
}
);
});
result->enqueue({result->address(), result}, make_any_tuple(atom("INIT")));
return result;
}
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> fun) {
return from([=](event_based_actor*) { fun(); });
}
event_based_actor::event_based_actor(actor_state st) : super(st, true) { }
resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
resume_result event_based_actor::resume(util::fiber*) {
CPPA_LOG_TRACE("id = " << id() << ", state = " << static_cast<int>(state()));
CPPA_REQUIRE( state() == actor_state::ready
|| state() == actor_state::pending);
scoped_self_setter sss{this};
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (exit_reason() == exit_reason::not_exited) {
......@@ -116,7 +112,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
m_bhvr_stack.cleanup();
on_exit();
CPPA_REQUIRE(next_job == nullptr);
next_job.swap(m_chained_actor);
return true;
};
CPPA_REQUIRE(next_job == nullptr);
......@@ -127,7 +122,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
next_job.swap(m_chained_actor);
set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) {
......@@ -137,7 +131,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
// interrupted by arriving message
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
next_job.swap(m_chained_actor);
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
break;
......@@ -155,7 +148,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
"mailbox can fetch more");
set_state(actor_state::ready);
CPPA_REQUIRE(m_chained_actor == nullptr);
next_job.swap(m_chained_actor);
}
}
else {
......
......@@ -58,7 +58,7 @@ group::module_ptr group::get_module(const std::string& module_name) {
return get_group_manager()->get_module(module_name);
}
group::subscription::subscription(const channel_ptr& s,
group::subscription::subscription(const channel& s,
const intrusive_ptr<group>& g)
: m_subscriber(s), m_group(g) { }
......@@ -88,8 +88,8 @@ const std::string& group::module_name() const {
}
struct group_nameserver : event_based_actor {
void init() {
become (
behavior make_behavior() override {
return (
on(atom("GET_GROUP"), arg_match) >> [](const std::string& name) {
return make_cow_tuple(atom("GROUP"), group::get("local", name));
},
......@@ -106,7 +106,7 @@ void publish_local_groups_at(std::uint16_t port, const char* addr) {
publish(gn, port, addr);
}
catch (std::exception&) {
gn->enqueue(nullptr, make_any_tuple(atom("SHUTDOWN")));
gn.enqueue({invalid_actor_addr, nullptr}, make_any_tuple(atom("SHUTDOWN")));
throw;
}
}
......
This diff is collapsed.
This diff is collapsed.
......@@ -72,11 +72,11 @@ class logging_impl : public logging {
void initialize() {
m_thread = thread([this] { (*this)(); });
log("TRACE", "logging", "run", __FILE__, __LINE__, nullptr, "ENTRY");
log("TRACE", "logging", "run", __FILE__, __LINE__, invalid_actor_addr, "ENTRY");
}
void destroy() {
log("TRACE", "logging", "run", __FILE__, __LINE__, nullptr, "EXIT");
log("TRACE", "logging", "run", __FILE__, __LINE__, invalid_actor_addr, "EXIT");
// an empty string means: shut down
m_queue.push_back(new log_event{0, ""});
m_thread.join();
......@@ -103,7 +103,7 @@ class logging_impl : public logging {
const char* function_name,
const char* c_full_file_name,
int line_num,
const actor_ptr& from,
actor_addr from,
const std::string& msg) {
string class_name = c_class_name;
replace_all(class_name, "::", ".");
......@@ -118,6 +118,7 @@ class logging_impl : public logging {
}
else file_name = move(full_file_name);
auto print_from = [&](ostream& oss) -> ostream& {
/*TODO:
if (!from) {
if (strcmp(c_class_name, "logging") == 0) oss << "logging";
else oss << "null";
......@@ -130,6 +131,7 @@ class logging_impl : public logging {
oss << from->id() << "@local";
# endif // CPPA_DEBUG_MODE
}
*/
return oss;
};
ostringstream line;
......@@ -158,7 +160,7 @@ logging::trace_helper::trace_helper(std::string class_name,
const char* fun_name,
const char* file_name,
int line_num,
actor_ptr ptr,
actor_addr ptr,
const std::string& msg)
: m_class(std::move(class_name)), m_fun_name(fun_name)
, m_file_name(file_name), m_line_num(line_num), m_self(std::move(ptr)) {
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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