Commit 5bd0f51d authored by Dominik Charousset's avatar Dominik Charousset

remodeled handling of sync. responses

parent 63eb3af5
...@@ -185,7 +185,7 @@ set(LIBCPPA_SRC ...@@ -185,7 +185,7 @@ set(LIBCPPA_SRC
src/ref_counted.cpp src/ref_counted.cpp
src/resumable.cpp src/resumable.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/response_handle.cpp src/response_promise.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduler.cpp src/scheduler.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
......
...@@ -108,7 +108,6 @@ cppa/match_expr.hpp ...@@ -108,7 +108,6 @@ cppa/match_expr.hpp
cppa/match_hint.hpp cppa/match_hint.hpp
cppa/memory_cached.hpp cppa/memory_cached.hpp
cppa/memory_managed.hpp cppa/memory_managed.hpp
cppa/message_future.hpp
cppa/message_header.hpp cppa/message_header.hpp
cppa/message_id.hpp cppa/message_id.hpp
cppa/message_priority.hpp cppa/message_priority.hpp
...@@ -137,7 +136,7 @@ cppa/primitive_variant.hpp ...@@ -137,7 +136,7 @@ cppa/primitive_variant.hpp
cppa/qtsupport/actor_widget_mixin.hpp cppa/qtsupport/actor_widget_mixin.hpp
cppa/ref_counted.hpp cppa/ref_counted.hpp
cppa/replies_to.hpp cppa/replies_to.hpp
cppa/response_handle.hpp cppa/response_promise.hpp
cppa/sb_actor.hpp cppa/sb_actor.hpp
cppa/scheduler.hpp cppa/scheduler.hpp
cppa/scoped_actor.hpp cppa/scoped_actor.hpp
...@@ -276,7 +275,7 @@ src/primitive_variant.cpp ...@@ -276,7 +275,7 @@ src/primitive_variant.cpp
src/protocol.cpp src/protocol.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/response_handle.cpp src/response_promise.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduler.cpp src/scheduler.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
......
...@@ -197,6 +197,8 @@ class abstract_actor : public abstract_channel { ...@@ -197,6 +197,8 @@ class abstract_actor : public abstract_channel {
// attached functors that are executed on cleanup // attached functors that are executed on cleanup
std::vector<attachable_ptr> m_attachables; std::vector<attachable_ptr> m_attachables;
protected:
// identifies the node this actor is running on // identifies the node this actor is running on
node_id_ptr m_node; node_id_ptr m_node;
......
...@@ -67,8 +67,10 @@ class actor_addr : util::comparable<actor_addr> ...@@ -67,8 +67,10 @@ class actor_addr : util::comparable<actor_addr>
actor_addr& operator=(const actor_addr&) = default; actor_addr& operator=(const actor_addr&) = default;
actor_addr(const actor&); actor_addr(const actor&);
actor_addr& operator=(const actor&);
actor_addr(const invalid_actor_addr_t&); actor_addr(const invalid_actor_addr_t&);
actor_addr operator=(const invalid_actor_addr_t&);
explicit operator bool() const; explicit operator bool() const;
bool operator!() const; bool operator!() const;
......
...@@ -112,6 +112,10 @@ class behavior { ...@@ -112,6 +112,10 @@ class behavior {
*/ */
behavior add_continuation(continuation_fun fun); behavior add_continuation(continuation_fun fun);
inline operator bool() const {
return static_cast<bool>(m_impl);
}
private: private:
impl_ptr m_impl; impl_ptr m_impl;
......
...@@ -31,7 +31,9 @@ ...@@ -31,7 +31,9 @@
#ifndef CPPA_BLOCKING_UNTYPED_ACTOR_HPP #ifndef CPPA_BLOCKING_UNTYPED_ACTOR_HPP
#define CPPA_BLOCKING_UNTYPED_ACTOR_HPP #define CPPA_BLOCKING_UNTYPED_ACTOR_HPP
#include "cppa/on.hpp"
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp" #include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
...@@ -45,6 +47,113 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> { ...@@ -45,6 +47,113 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
public: public:
class response_future {
public:
response_future() = delete;
void await(behavior&);
inline void await(behavior&& bhvr) {
behavior arg{std::move(bhvr)};
await(arg);
}
/**
* @brief Blocks until the response arrives and then executes @p mexpr.
*/
template<typename... Cs, typename... Ts>
void await(const match_expr<Cs...>& arg, const Ts&... args) {
await(match_expr_convert(arg, args...));
}
/**
* @brief Blocks until the response arrives and then executes @p @p fun,
* calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
await(behavior{(on_arg_match >> std::move(fs))...});
}
/**
* @brief Returns the awaited response ID.
*/
inline const message_id& id() const { return m_mid; }
response_future(const response_future&) = default;
response_future& operator=(const response_future&) = default;
inline response_future(const message_id& from,
blocking_untyped_actor* self)
: m_mid(from), m_self(self) { }
private:
message_id m_mid;
blocking_untyped_actor* m_self;
};
class sync_receive_helper {
public:
inline sync_receive_helper(const response_future& mf) : m_mf(mf) { }
template<typename... Ts>
inline void operator()(Ts&&... args) {
m_mf.await(std::forward<Ts>(args)...);
}
private:
response_future m_mf;
};
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future sync_send_tuple(const actor& dest, any_tuple what);
response_future timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_future sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(dest, make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
response_future timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest, make_any_tuple(std::forward<Ts>(what)...));
}
typedef std::chrono::high_resolution_clock::time_point timeout_type; typedef std::chrono::high_resolution_clock::time_point timeout_type;
struct receive_while_helper { struct receive_while_helper {
...@@ -162,9 +271,9 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> { ...@@ -162,9 +271,9 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
* @param handle A future for a synchronous response. * @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor * @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle * already received the response for @p handle
* @relates message_future * @relates response_future
*/ */
inline sync_receive_helper receive_response(const message_future& f) { inline sync_receive_helper receive_response(const response_future& f) {
return {f}; return {f};
} }
...@@ -203,7 +312,11 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> { ...@@ -203,7 +312,11 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
dequeue(tmp); dequeue(tmp);
} }
virtual void dequeue(behavior& bhvr) = 0; inline void dequeue(behavior& bhvr) {
dequeue_response(bhvr, message_id::invalid);
}
virtual void dequeue_response(behavior& bhvr, message_id mid) = 0;
virtual mailbox_element* dequeue() = 0; virtual mailbox_element* dequeue() = 0;
......
...@@ -59,8 +59,7 @@ ...@@ -59,8 +59,7 @@
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/untyped_actor.hpp" #include "cppa/untyped_actor.hpp"
#include "cppa/abstract_actor.hpp" #include "cppa/abstract_actor.hpp"
#include "cppa/message_future.hpp" #include "cppa/response_promise.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/blocking_untyped_actor.hpp" #include "cppa/blocking_untyped_actor.hpp"
#include "cppa/util/type_traits.hpp" #include "cppa/util/type_traits.hpp"
......
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
#include <type_traits> #include <type_traits>
#include "cppa/logging.hpp"
#include "cppa/resumable.hpp" #include "cppa/resumable.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/blocking_untyped_actor.hpp" #include "cppa/blocking_untyped_actor.hpp"
...@@ -38,6 +39,9 @@ class proper_actor_base : public ResumePolicy::template mixin<Base> { ...@@ -38,6 +39,9 @@ class proper_actor_base : public ResumePolicy::template mixin<Base> {
proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) { } proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) { }
void enqueue(const message_header& hdr, any_tuple msg) override { void enqueue(const message_header& hdr, any_tuple msg) override {
CPPA_PUSH_AID(this->id());
CPPA_LOG_DEBUG(CPPA_TARG(hdr, to_string)
<< ", " << CPPA_TARG(msg, to_string));
m_scheduling_policy.enqueue(this, hdr, msg); m_scheduling_policy.enqueue(this, hdr, msg);
} }
...@@ -97,12 +101,14 @@ class proper_actor : public proper_actor_base<Base, SchedulingPolicy, ...@@ -97,12 +101,14 @@ class proper_actor : public proper_actor_base<Base, SchedulingPolicy,
detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; } detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; }
inline void launch() { inline void launch() {
this->bhvr_stack().push_back(this->make_behavior()); auto bhvr = this->make_behavior();
if (bhvr) this->bhvr_stack().push_back(std::move(bhvr));
this->m_scheduling_policy.launch(this); this->m_scheduling_policy.launch(this);
} }
bool invoke(mailbox_element* msg) { bool invoke(mailbox_element* msg) {
return this->m_invoke_policy.invoke(this, msg, bhvr_stack().back()); return this->m_invoke_policy.invoke(this, msg, bhvr_stack().back(),
bhvr_stack().back_id());
} }
}; };
...@@ -129,7 +135,7 @@ class proper_actor< ...@@ -129,7 +135,7 @@ class proper_actor<
this->m_scheduling_policy.launch(this); this->m_scheduling_policy.launch(this);
} }
void dequeue(behavior& bhvr) override { void dequeue_response(behavior& bhvr, message_id mid) override {
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid()) {
auto tout = auto tout =
this->m_scheduling_policy.init_timeout(this, bhvr.timeout()); this->m_scheduling_policy.init_timeout(this, bhvr.timeout());
...@@ -143,14 +149,14 @@ class proper_actor< ...@@ -143,14 +149,14 @@ class proper_actor<
// must not return nullptr, because await_data guarantees // must not return nullptr, because await_data guarantees
// at least one message in our mailbox // at least one message in our mailbox
CPPA_REQUIRE(msg != nullptr); CPPA_REQUIRE(msg != nullptr);
done = this->m_invoke_policy.invoke(this, msg, bhvr); done = this->m_invoke_policy.invoke(this, msg, bhvr, mid);
} }
} }
} else { } else {
for (;;) { for (;;) {
auto msg = this->m_priority_policy.next_message(this); auto msg = this->m_priority_policy.next_message(this);
while (msg) { while (msg) {
if (this->m_invoke_policy.invoke(this, msg, bhvr)) { if (this->m_invoke_policy.invoke(this, msg, bhvr, mid)) {
// we're done // we're done
return; return;
} }
......
...@@ -48,11 +48,10 @@ ...@@ -48,11 +48,10 @@
#include "cppa/typed_actor.hpp" #include "cppa/typed_actor.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/memory_cached.hpp" #include "cppa/memory_cached.hpp"
#include "cppa/message_future.hpp"
#include "cppa/message_header.hpp" #include "cppa/message_header.hpp"
#include "cppa/abstract_actor.hpp" #include "cppa/abstract_actor.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/response_handle.hpp" #include "cppa/response_promise.hpp"
#include "cppa/message_priority.hpp" #include "cppa/message_priority.hpp"
#include "cppa/partial_function.hpp" #include "cppa/partial_function.hpp"
...@@ -66,7 +65,7 @@ namespace cppa { ...@@ -66,7 +65,7 @@ namespace cppa {
// forward declarations // forward declarations
class scheduler; class scheduler;
class message_future; class response_future;
class local_scheduler; class local_scheduler;
class sync_handle_helper; class sync_handle_helper;
...@@ -149,45 +148,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -149,45 +148,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void send_exit(const actor_addr& whom, std::uint32_t reason); void send_exit(const actor_addr& whom, std::uint32_t reason);
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
message_future sync_send_tuple(const actor& dest, any_tuple what);
message_future timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline message_future sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(dest, make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
message_future timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest, make_any_tuple(std::forward<Ts>(what)...));
}
/** /**
* @brief Sends a message to @p whom that is delayed by @p rel_time. * @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message. * @param whom Receiver of the message.
...@@ -307,10 +267,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -307,10 +267,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
std::vector<group_ptr> joined_groups() const; std::vector<group_ptr> joined_groups() const;
/** /**
* @brief Creates a {@link response_handle} to allow actors to response * @brief Creates a {@link response_promise} to allow actors to response
* to a request later on. * to a request later on.
*/ */
response_handle make_response_handle(); response_promise make_response_promise();
/** /**
* @brief Sets the handler for unexpected synchronous response messages. * @brief Sets the handler for unexpected synchronous response messages.
......
This diff is collapsed.
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_MESSAGE_FUTURE_HPP
#define CPPA_MESSAGE_FUTURE_HPP
#include <cstdint>
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/partial_function.hpp"
namespace cppa {
class local_actor;
namespace detail { struct optional_any_tuple_visitor; }
/**
* @brief Provides the @p continue_with member function as used in
* <tt>sync_send(...).then(...).continue_with(...)</tt>.
*/
class continue_helper {
public:
typedef int message_id_wrapper_tag;
inline continue_helper(message_id mid) : m_mid(mid) { }
template<typename F>
continue_helper& continue_with(F) {
//FIXME
throw std::logic_error("not implemented yet");
}
inline message_id get_message_id() const {
return m_mid;
}
private:
message_id m_mid;
local_actor* self;
};
/**
* @brief Represents the result of a synchronous send.
*/
class message_future {
public:
message_future() = delete;
inline continue_helper then(const behavior&) {
//FIXME
return message_id{};
}
inline continue_helper await(const behavior&) {
//FIXME
return message_id{};
}
/**
* @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) {
return then(match_expr_convert(arg, args...));
}
/**
* @brief Blocks until the response arrives and then executes @p mexpr.
*/
template<typename... Cs, typename... Ts>
void await(const match_expr<Cs...>& arg, const Ts&... args) {
await(match_expr_convert(arg, args...));
}
/**
* @brief Sets @p fun as event-handler for the response message, calls
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<
util::all_callable<Fs...>::value,
continue_helper
>::type
then(Fs... fs) {
return then(partial_function{(on_arg_match >> std::move(fs))...});
}
/**
* @brief Blocks until the response arrives and then executes @p @p fun,
* calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
await(partial_function{(on_arg_match >> std::move(fs))...});
}
/**
* @brief Returns the awaited response ID.
*/
inline const message_id& id() const { return m_mid; }
message_future(const message_future&) = default;
message_future& operator=(const message_future&) = default;
inline message_future(const message_id& from) : m_mid(from) { }
private:
message_id m_mid;
local_actor* self;
inline void check_consistency() { }
};
class sync_handle_helper {
public:
inline sync_handle_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Ts>
inline continue_helper operator()(Ts&&... args) {
return m_mf.then(std::forward<Ts>(args)...);
}
private:
message_future m_mf;
};
class sync_receive_helper {
public:
inline sync_receive_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Ts>
inline void operator()(Ts&&... args) {
m_mf.await(std::forward<Ts>(args)...);
}
private:
message_future m_mf;
};
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_handle_helper handle_response(const message_future& f) {
return {f};
}
} // namespace cppa
#endif // CPPA_MESSAGE_FUTURE_HPP
...@@ -31,9 +31,11 @@ ...@@ -31,9 +31,11 @@
#ifndef PRIORITY_HPP #ifndef PRIORITY_HPP
#define PRIORITY_HPP #define PRIORITY_HPP
#include <cstdint>
namespace cppa { namespace cppa {
enum class message_priority { enum class message_priority : std::uint32_t {
normal, normal,
high high
}; };
......
...@@ -139,7 +139,7 @@ class actor_facade<Ret(Args...)> : public abstract_actor { ...@@ -139,7 +139,7 @@ class actor_facade<Ret(Args...)> : public abstract_actor {
util::int_list<Is...>) { util::int_list<Is...>) {
auto opt = m_map_args(std::move(msg)); auto opt = m_map_args(std::move(msg));
if (opt) { if (opt) {
response_handle handle{ this, sender, id.response_id() }; response_promise handle{ this, sender, id.response_id() };
size_t ret_size = std::accumulate(m_global_dimensions.begin(), size_t ret_size = std::accumulate(m_global_dimensions.begin(),
m_global_dimensions.end(), 1, m_global_dimensions.end(), 1,
std::multiplies<size_t>{}); std::multiplies<size_t>{});
......
...@@ -40,7 +40,7 @@ ...@@ -40,7 +40,7 @@
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/opencl/global.hpp" #include "cppa/opencl/global.hpp"
#include "cppa/abstract_actor.hpp" #include "cppa/abstract_actor.hpp"
#include "cppa/response_handle.hpp" #include "cppa/response_promise.hpp"
#include "cppa/opencl/smart_ptr.hpp" #include "cppa/opencl/smart_ptr.hpp"
#include "cppa/util/scope_guard.hpp" #include "cppa/util/scope_guard.hpp"
...@@ -51,7 +51,7 @@ class command : public ref_counted { ...@@ -51,7 +51,7 @@ class command : public ref_counted {
public: public:
command(response_handle handle, command(response_promise handle,
intrusive_ptr<T> actor_facade, intrusive_ptr<T> actor_facade,
std::vector<mem_ptr> arguments) std::vector<mem_ptr> arguments)
: m_number_of_values(std::accumulate(actor_facade->m_global_dimensions.begin(), : m_number_of_values(std::accumulate(actor_facade->m_global_dimensions.begin(),
...@@ -107,7 +107,7 @@ class command : public ref_counted { ...@@ -107,7 +107,7 @@ class command : public ref_counted {
private: private:
int m_number_of_values; int m_number_of_values;
response_handle m_handle; response_promise m_handle;
intrusive_ptr<T> m_actor_facade; intrusive_ptr<T> m_actor_facade;
event_ptr m_kernel_event; event_ptr m_kernel_event;
command_queue_ptr m_queue; command_queue_ptr m_queue;
......
...@@ -239,6 +239,16 @@ class optional<T&> { ...@@ -239,6 +239,16 @@ class optional<T&> {
return *m_value; return *m_value;
} }
inline T* operator->() {
CPPA_REQUIRE(valid());
return m_value;
}
inline const T* operator->() const {
CPPA_REQUIRE(valid());
return m_value;
}
inline T& get() { inline T& get() {
CPPA_REQUIRE(valid()); CPPA_REQUIRE(valid());
return *m_value; return *m_value;
......
...@@ -71,6 +71,7 @@ class context_switching_resume { ...@@ -71,6 +71,7 @@ class context_switching_resume {
resumable::resume_result resume(util::fiber* from) override { resumable::resume_result resume(util::fiber* from) override {
CPPA_REQUIRE(from != nullptr); CPPA_REQUIRE(from != nullptr);
CPPA_PUSH_AID(this->id());
using namespace detail; using namespace detail;
for (;;) { for (;;) {
switch (call(&m_fiber, from)) { switch (call(&m_fiber, from)) {
......
...@@ -65,6 +65,7 @@ class event_based_resume { ...@@ -65,6 +65,7 @@ class event_based_resume {
<< ", state = " << static_cast<int>(this->state())); << ", state = " << static_cast<int>(this->state()));
CPPA_REQUIRE( this->state() == actor_state::ready CPPA_REQUIRE( this->state() == actor_state::ready
|| this->state() == actor_state::pending); || this->state() == actor_state::pending);
CPPA_PUSH_AID(this->id());
auto done_cb = [&]() -> bool { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
if (this->exit_reason() == exit_reason::not_exited) { if (this->exit_reason() == exit_reason::not_exited) {
...@@ -89,7 +90,7 @@ class event_based_resume { ...@@ -89,7 +90,7 @@ class event_based_resume {
for (auto e = this->m_mailbox.try_pop(); ; e = this->m_mailbox.try_pop()) { for (auto e = this->m_mailbox.try_pop(); ; e = this->m_mailbox.try_pop()) {
//e = m_mailbox.try_pop(); //e = m_mailbox.try_pop();
if (e == nullptr) { if (e == nullptr) {
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block"); CPPA_LOG_DEBUG("no more element in mailbox; going to block");
this->set_state(actor_state::about_to_block); this->set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) { if (this->m_mailbox.can_fetch_more() == false) {
...@@ -103,28 +104,24 @@ class event_based_resume { ...@@ -103,28 +104,24 @@ class event_based_resume {
"arriving message"); "arriving message");
break; break;
case actor_state::blocked: case actor_state::blocked:
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked"); CPPA_LOG_DEBUG("set state successfully to blocked");
// done setting actor to blocked // done setting actor to blocked
return resumable::resume_later; return resumable::resume_later;
default: default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid state"); CPPA_LOG_ERROR("invalid state");
CPPA_CRITICAL("invalid state"); CPPA_CRITICAL("invalid state");
}; };
} }
else { else {
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: " CPPA_LOG_DEBUG("switched back to ready: "
"mailbox can fetch more"); "mailbox can fetch more");
this->set_state(actor_state::ready); this->set_state(actor_state::ready);
} }
} }
else { else {
if (this->invoke(e)) { if (this->invoke(e)) {
CPPA_LOG_DEBUG_IF(m_chained_actor,
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (this->bhvr_stack().empty() && done_cb()) { if (this->bhvr_stack().empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty"); CPPA_LOG_DEBUG("behavior stack empty");
return resume_result::done; return resume_result::done;
} }
this->bhvr_stack().cleanup(); this->bhvr_stack().cleanup();
......
...@@ -44,6 +44,7 @@ ...@@ -44,6 +44,7 @@
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/partial_function.hpp" #include "cppa/partial_function.hpp"
#include "cppa/response_promise.hpp"
#include "cppa/util/dptr.hpp" #include "cppa/util/dptr.hpp"
#include "cppa/detail/memory.hpp" #include "cppa/detail/memory.hpp"
...@@ -91,7 +92,7 @@ class invoke_policy { ...@@ -91,7 +92,7 @@ class invoke_policy {
auto i = m_cache.begin(); auto i = m_cache.begin();
auto e = m_cache.end(); auto e = m_cache.end();
while (i != e) { while (i != e) {
switch (this->handle_message(self, i->get(), fun, awaited_response)) { switch (handle_message(self, i->get(), fun, awaited_response)) {
case hm_msg_handled: { case hm_msg_handled: {
m_cache.erase(i); m_cache.erase(i);
return true; return true;
...@@ -123,8 +124,7 @@ class invoke_policy { ...@@ -123,8 +124,7 @@ class invoke_policy {
Fun& fun, Fun& fun,
message_id awaited_response = message_id()) { message_id awaited_response = message_id()) {
smart_pointer node(node_ptr); smart_pointer node(node_ptr);
switch (this->handle_message(self, node.get(), fun, switch (handle_message(self, node.get(), fun, awaited_response)) {
awaited_response)) {
case hm_msg_handled: { case hm_msg_handled: {
return true; return true;
} }
...@@ -283,12 +283,12 @@ class invoke_policy { ...@@ -283,12 +283,12 @@ class invoke_policy {
public: public:
template<class Actor> template<class Actor>
inline response_handle fetch_response_handle(Actor* cl, int) { inline response_promise fetch_response_promise(Actor* cl, int) {
return cl->make_response_handle(); return cl->make_response_promise();
} }
template<class Actor> template<class Actor>
inline response_handle fetch_response_handle(Actor*, response_handle& hdl) { inline response_promise fetch_response_promise(Actor*, response_promise& hdl) {
return std::move(hdl); return std::move(hdl);
} }
...@@ -308,23 +308,24 @@ class invoke_policy { ...@@ -308,23 +308,24 @@ class invoke_policy {
// make sure synchronous requests // make sure synchronous requests
// always receive a response // always receive a response
if (mid.is_request() && !mid.is_answered()) { if (mid.is_request() && !mid.is_answered()) {
CPPA_LOGMF(CPPA_WARNING, self, CPPA_LOG_WARNING("actor with ID " << self->id()
"actor with ID " << self->id()
<< " did not reply to a " << " did not reply to a "
"synchronous request message"); "synchronous request message");
auto fhdl = fetch_response_handle(self, hdl); auto fhdl = fetch_response_promise(self, hdl);
if (fhdl.valid()) fhdl.apply(make_any_tuple(atom("VOID"))); if (fhdl) fhdl.deliver(make_any_tuple(atom("VOID")));
} }
} else { } else {
if ( detail::matches<atom_value, std::uint64_t>(*res) if ( detail::matches<atom_value, std::uint64_t>(*res)
&& res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) { && res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) {
CPPA_LOG_DEBUG("message handler returned a "
"message id wrapper");
auto id = res->template get_as<std::uint64_t>(1); auto id = res->template get_as<std::uint64_t>(1);
auto msg_id = message_id::from_integer_value(id); auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->sync_handler(msg_id); auto ref_opt = self->sync_handler(msg_id);
// calls self->response_handle() if hdl is a dummy // calls self->response_promise() if hdl is a dummy
// argument, forwards hdl otherwise to reply to the // argument, forwards hdl otherwise to reply to the
// original request message // original request message
auto fhdl = fetch_response_handle(self, hdl); auto fhdl = fetch_response_promise(self, hdl);
if (ref_opt) { if (ref_opt) {
auto& ref = *ref_opt; auto& ref = *ref_opt;
// copy original behavior // copy original behavior
...@@ -356,8 +357,9 @@ class invoke_policy { ...@@ -356,8 +357,9 @@ class invoke_policy {
res->reset(); res->reset();
} else { } else {
// respond by using the result of 'fun' // respond by using the result of 'fun'
auto fhdl = fetch_response_handle(self, hdl); CPPA_LOG_DEBUG("respond via response_promise");
if (fhdl.valid()) fhdl.apply(std::move(*res)); auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) fhdl.deliver(std::move(*res));
} }
} }
return res; return res;
...@@ -384,26 +386,29 @@ class invoke_policy { ...@@ -384,26 +386,29 @@ class invoke_policy {
} }
switch (this->filter_msg(self, node)) { switch (this->filter_msg(self, node)) {
default: { default: {
CPPA_LOG_ERROR("illegal filter result");
CPPA_CRITICAL("illegal filter result"); CPPA_CRITICAL("illegal filter result");
} }
case normal_exit_signal: { case normal_exit_signal: {
CPPA_LOGMF(CPPA_DEBUG, self, "dropped normal exit signal"); CPPA_LOG_DEBUG("dropped normal exit signal");
return hm_drop_msg; return hm_drop_msg;
} }
case expired_sync_response: { case expired_sync_response: {
CPPA_LOGMF(CPPA_DEBUG, self, "dropped expired sync response"); CPPA_LOG_DEBUG("dropped expired sync response");
return hm_drop_msg; return hm_drop_msg;
} }
case expired_timeout_message: { case expired_timeout_message: {
CPPA_LOGMF(CPPA_DEBUG, self, "dropped expired timeout message"); CPPA_LOG_DEBUG("dropped expired timeout message");
return hm_drop_msg; return hm_drop_msg;
} }
case non_normal_exit_signal: { case non_normal_exit_signal: {
CPPA_LOG_DEBUG("handled non-normal exit signal");
// this message was handled // this message was handled
// by calling self->quit(...) // by calling self->quit(...)
return hm_msg_handled; return hm_msg_handled;
} }
case timeout_message: { case timeout_message: {
CPPA_LOG_DEBUG("handle timeout message");
dptr()->handle_timeout(self, fun); dptr()->handle_timeout(self, fun);
if (awaited_response.valid()) { if (awaited_response.valid()) {
self->mark_arrived(awaited_response); self->mark_arrived(awaited_response);
...@@ -416,6 +421,10 @@ class invoke_policy { ...@@ -416,6 +421,10 @@ class invoke_policy {
// fall through // fall through
} }
case sync_response: { case sync_response: {
CPPA_LOG_DEBUG("handle as synchronous response: "
<< CPPA_TARG(node->msg, to_string) << ", "
<< CPPA_MARG(node->mid, integer_value) << ", "
<< CPPA_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node->mid == awaited_response) { if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = dptr()->hm_begin(self, node); auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, auto res = invoke_fun(self,
...@@ -423,10 +432,8 @@ class invoke_policy { ...@@ -423,10 +432,8 @@ class invoke_policy {
node->mid, node->mid,
fun); fun);
if (!res && handle_sync_failure_on_mismatch) { if (!res && handle_sync_failure_on_mismatch) {
CPPA_LOGMF(CPPA_WARNING, CPPA_LOG_WARNING("sync failure occured in actor "
self, << "with ID " << self->id());
"sync failure occured in actor with ID "
<< self->id());
self->handle_sync_failure(); self->handle_sync_failure();
} }
self->mark_arrived(awaited_response); self->mark_arrived(awaited_response);
...@@ -437,6 +444,8 @@ class invoke_policy { ...@@ -437,6 +444,8 @@ class invoke_policy {
return hm_cache_msg; return hm_cache_msg;
} }
case ordinary_message: { case ordinary_message: {
CPPA_LOG_DEBUG("handle as ordinary message: "
<< CPPA_TARG(node->msg, to_string));
if (!awaited_response.valid()) { if (!awaited_response.valid()) {
auto previous_node = dptr()->hm_begin(self, node); auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, auto res = invoke_fun(self,
......
...@@ -36,11 +36,18 @@ ...@@ -36,11 +36,18 @@
#include <chrono> #include <chrono>
#include <condition_variable> #include <condition_variable>
#include "cppa/logging.hpp"
#include "cppa/singletons.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/util/fiber.hpp" #include "cppa/util/fiber.hpp"
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
#include "cppa/util/scope_guard.hpp"
#include "cppa/policy/scheduling_policy.hpp" #include "cppa/policy/scheduling_policy.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/sync_request_bouncer.hpp" #include "cppa/detail/sync_request_bouncer.hpp"
#include "cppa/intrusive/single_reader_queue.hpp" #include "cppa/intrusive/single_reader_queue.hpp"
...@@ -92,20 +99,17 @@ class no_scheduling { ...@@ -92,20 +99,17 @@ class no_scheduling {
template<class Actor> template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) { void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
std::cout << "enqueue\n";
auto ptr = self->new_mailbox_element(hdr, std::move(msg)); auto ptr = self->new_mailbox_element(hdr, std::move(msg));
switch (self->mailbox().enqueue(ptr)) { switch (self->mailbox().enqueue(ptr)) {
default: default:
std::cout << "enqueue: default case (do nothing)\n";
break; break;
case intrusive::first_enqueued: { case intrusive::first_enqueued: {
std::cout << "enqueue: first enqueue -> notify\n";
lock_type guard(m_mtx); lock_type guard(m_mtx);
self->set_state(actor_state::ready);
m_cv.notify_one(); m_cv.notify_one();
break; break;
} }
case intrusive::queue_closed: case intrusive::queue_closed:
std::cout << "enqueue: mailbox closed!\n";
if (hdr.id.valid()) { if (hdr.id.valid()) {
detail::sync_request_bouncer f{self->exit_reason()}; detail::sync_request_bouncer f{self->exit_reason()};
f(hdr.sender, hdr.id); f(hdr.sender, hdr.id);
...@@ -116,15 +120,27 @@ std::cout << "enqueue: mailbox closed!\n"; ...@@ -116,15 +120,27 @@ std::cout << "enqueue: mailbox closed!\n";
template<class Actor> template<class Actor>
void launch(Actor* self) { void launch(Actor* self) {
CPPA_PUSH_AID(self->id());
get_actor_registry()->inc_running();
std::thread([=] { std::thread([=] {
CPPA_PUSH_AID(self->id());
CPPA_LOG_TRACE("");
auto guard = util::make_scope_guard([] {
get_actor_registry()->dec_running();
});
util::fiber fself; util::fiber fself;
auto rr = resumable::resume_later; for (;;) {
while (rr != resumable::done) {
std::cout << "before await_data\n";
await_data(self); await_data(self);
std::cout << "before resume\n"; if (self->resume(&fself) == resumable::done) {
rr = self->resume(&fself); CPPA_LOG_DEBUG("resume returned resumable::done");
std::cout << "after resume\n"; self->planned_exit_reason(exit_reason::normal);
}
auto per = self->planned_exit_reason();
if (per != exit_reason::not_exited) {
CPPA_LOG_DEBUG("planned exit reason: " << per);
self->cleanup(per);
return;
}
} }
}).detach(); }).detach();
} }
......
...@@ -38,52 +38,34 @@ ...@@ -38,52 +38,34 @@
namespace cppa { namespace cppa {
/** /**
* @brief Denotes an outstanding response. * @brief A response promise can be used to deliver a uniquely identifiable
* response message from the server (i.e. receiver of the request)
* to the client (i.e. the sender of the request).
*/ */
class response_handle { class response_promise {
public: public:
response_handle() = default; response_promise() = default;
response_handle(response_handle&&) = default; response_promise(response_promise&&) = default;
response_handle(const response_handle&) = default; response_promise(const response_promise&) = default;
response_handle& operator=(response_handle&&) = default; response_promise& operator=(response_promise&&) = default;
response_handle& operator=(const response_handle&) = default; response_promise& operator=(const response_promise&) = default;
response_handle(const actor_addr& from, response_promise(const actor_addr& from,
const actor_addr& to, const actor_addr& to,
const message_id& response_id); const message_id& response_id);
/** /**
* @brief Queries whether response message is still outstanding. * @brief Queries whether this promise is still valid, i.e., no response
* was yet delivered to the client.
*/ */
bool valid() const; explicit operator bool() const;
/**
* @brief Queries whether this is a response
* handle for a synchronous request.
*/
bool synchronous() const;
/** /**
* @brief Sends @p response_message and invalidates this handle afterwards. * @brief Sends @p response_message and invalidates this handle afterwards.
*/ */
void apply(any_tuple response_message) const; void deliver(any_tuple response_message);
/**
* @brief Returns the message id for the response message.
*/
inline const message_id& response_id() const { return m_id; }
/**
* @brief Returns the actor that is going send the response message.
*/
inline const actor_addr& sender() const { return m_from; }
/**
* @brief Returns the actor that is waiting for the response message.
*/
inline const actor_addr& receiver() const { return m_to; }
private: private:
......
...@@ -40,10 +40,12 @@ ...@@ -40,10 +40,12 @@
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/channel.hpp" #include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/cow_tuple.hpp" #include "cppa/cow_tuple.hpp"
#include "cppa/resumable.hpp" #include "cppa/resumable.hpp"
#include "cppa/attachable.hpp" #include "cppa/attachable.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/message_header.hpp"
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
......
...@@ -70,6 +70,7 @@ class scoped_actor { ...@@ -70,6 +70,7 @@ class scoped_actor {
private: private:
actor_id m_prev;
intrusive_ptr<blocking_untyped_actor> m_self; intrusive_ptr<blocking_untyped_actor> m_self;
}; };
......
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/message_header.hpp" #include "cppa/message_header.hpp"
#include "cppa/message_future.hpp"
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
...@@ -126,7 +125,7 @@ inline void send_as(actor_ptr from, channel_destination dest, Ts&&... what) { ...@@ -126,7 +125,7 @@ inline void send_as(actor_ptr from, channel_destination dest, Ts&&... what) {
* message cannot be received by another actor. * message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt> * @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/ */
message_future sync_send_tuple(actor_destination dest, any_tuple what); response_future sync_send_tuple(actor_destination dest, any_tuple what);
/** /**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom. * @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
...@@ -139,7 +138,7 @@ message_future sync_send_tuple(actor_destination dest, any_tuple what); ...@@ -139,7 +138,7 @@ message_future sync_send_tuple(actor_destination dest, any_tuple what);
* @throws std::invalid_argument if <tt>whom == nullptr</tt> * @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/ */
template<typename... Ts> template<typename... Ts>
inline message_future sync_send(actor_destination dest, Ts&&... what) { inline response_future sync_send(actor_destination dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(std::move(dest), return sync_send_tuple(std::move(dest),
make_any_tuple(std::forward<Ts>(what)...)); make_any_tuple(std::forward<Ts>(what)...));
...@@ -156,7 +155,7 @@ inline message_future sync_send(actor_destination dest, Ts&&... what) { ...@@ -156,7 +155,7 @@ inline message_future sync_send(actor_destination dest, Ts&&... what) {
* @throws std::invalid_argument if <tt>whom == nullptr</tt> * @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/ */
template<typename... Signatures, typename... Ts> template<typename... Signatures, typename... Ts>
typed_message_future< typed_response_future<
typename detail::deduce_output_type< typename detail::deduce_output_type<
util::type_list<Signatures...>, util::type_list<Signatures...>,
util::type_list< util::type_list<
...@@ -216,7 +215,7 @@ void delayed_reply_tuple(const util::duration& rel_time, any_tuple data); ...@@ -216,7 +215,7 @@ void delayed_reply_tuple(const util::duration& rel_time, any_tuple data);
* message cannot be received by another actor. * message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt> * @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/ */
message_future timed_sync_send_tuple(actor_destination dest, response_future timed_sync_send_tuple(actor_destination dest,
const util::duration& rtime, const util::duration& rtime,
any_tuple what); any_tuple what);
...@@ -235,7 +234,7 @@ message_future timed_sync_send_tuple(actor_destination dest, ...@@ -235,7 +234,7 @@ message_future timed_sync_send_tuple(actor_destination dest,
* @throws std::invalid_argument if <tt>whom == nullptr</tt> * @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/ */
template<typename... Ts> template<typename... Ts>
message_future timed_sync_send(actor_destination whom, response_future timed_sync_send(actor_destination whom,
const util::duration& rtime, const util::duration& rtime,
Ts&&... what) { Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
...@@ -265,7 +264,7 @@ CPPA_DEPRECATED inline void reply(Ts&&... what) { ...@@ -265,7 +264,7 @@ CPPA_DEPRECATED inline void reply(Ts&&... what) {
* @brief Sends a message as reply to @p handle. * @brief Sends a message as reply to @p handle.
*/ */
template<typename... Ts> template<typename... Ts>
inline void reply_to(const response_handle& handle, Ts&&... what) { inline void reply_to(const response_promise& handle, Ts&&... what) {
if (handle.valid()) { if (handle.valid()) {
handle.apply(make_any_tuple(std::forward<Ts>(what)...)); handle.apply(make_any_tuple(std::forward<Ts>(what)...));
} }
...@@ -276,7 +275,7 @@ inline void reply_to(const response_handle& handle, Ts&&... what) { ...@@ -276,7 +275,7 @@ inline void reply_to(const response_handle& handle, Ts&&... what) {
* @param handle Identifies a previously received request. * @param handle Identifies a previously received request.
* @param what Response message. * @param what Response message.
*/ */
inline void reply_tuple_to(const response_handle& handle, any_tuple what) { inline void reply_tuple_to(const response_promise& handle, any_tuple what) {
handle.apply(std::move(what)); handle.apply(std::move(what));
} }
......
...@@ -33,7 +33,6 @@ ...@@ -33,7 +33,6 @@
#include "cppa/replies_to.hpp" #include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp" #include "cppa/typed_behavior.hpp"
#include "cppa/message_future.hpp"
#include "cppa/untyped_actor.hpp" #include "cppa/untyped_actor.hpp"
#include "cppa/detail/typed_actor_util.hpp" #include "cppa/detail/typed_actor_util.hpp"
......
...@@ -31,13 +31,46 @@ ...@@ -31,13 +31,46 @@
#ifndef CPPA_UNTYPED_ACTOR_HPP #ifndef CPPA_UNTYPED_ACTOR_HPP
#define CPPA_UNTYPED_ACTOR_HPP #define CPPA_UNTYPED_ACTOR_HPP
#include "cppa/on.hpp"
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/logging.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp" #include "cppa/mailbox_based.hpp"
#include "cppa/behavior_stack_based.hpp" #include "cppa/behavior_stack_based.hpp"
namespace cppa { namespace cppa {
class untyped_actor;
class continue_helper {
public:
typedef int message_id_wrapper_tag;
inline continue_helper(message_id mid, untyped_actor* self)
: m_mid(mid), m_self(self) { }
template<typename F>
continue_helper& continue_with(F fun) {
return continue_with(behavior::continuation_fun{partial_function{
on(any_vals, arg_match) >> fun
}});
}
continue_helper& continue_with(behavior::continuation_fun fun);
message_id get_message_id() const {
return m_mid;
}
private:
message_id m_mid;
untyped_actor* m_self;
};
/** /**
* @extends local_actor * @extends local_actor
*/ */
...@@ -50,6 +83,97 @@ class untyped_actor : public extend<local_actor>::with<mailbox_based, ...@@ -50,6 +83,97 @@ class untyped_actor : public extend<local_actor>::with<mailbox_based,
void forward_to(const actor& other); void forward_to(const actor& other);
public:
class response_future {
public:
response_future() = delete;
inline continue_helper then(behavior bhvr) {
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
return {m_mid, m_self};
}
/**
* @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) {
return then(match_expr_convert(arg, args...));
}
/**
* @brief Sets @p fun as event-handler for the response message, calls
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<
util::all_callable<Fs...>::value,
continue_helper
>::type
then(Fs... fs) {
return then(behavior{(on_arg_match >> std::move(fs))...});
}
response_future(const response_future&) = default;
response_future& operator=(const response_future&) = default;
inline response_future(const message_id& from, untyped_actor* self)
: m_mid(from), m_self(self) { }
private:
message_id m_mid;
untyped_actor* m_self;
inline void check_consistency() { }
};
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future sync_send_tuple(const actor& dest, any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_future sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(dest, make_any_tuple(std::forward<Ts>(what)...));
}
response_future timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what);
template<typename... Ts>
response_future timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest, make_any_tuple(std::forward<Ts>(what)...));
}
}; };
} // namespace cppa } // namespace cppa
......
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include <atomic> #include <atomic>
#include <stdexcept> #include <stdexcept>
#include "cppa/atom.hpp"
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
......
...@@ -68,4 +68,14 @@ intptr_t actor_addr::compare(const local_actor* other) const { ...@@ -68,4 +68,14 @@ intptr_t actor_addr::compare(const local_actor* other) const {
return compare_impl(m_ops.m_ptr.get(), other); return compare_impl(m_ops.m_ptr.get(), other);
} }
actor_addr& actor_addr::operator=(const actor& other) {
m_ops.m_ptr = detail::raw_access::get(other);
return *this;
}
actor_addr actor_addr::operator=(const invalid_actor_addr_t&) {
m_ops.m_ptr.reset();
return *this;
}
} // namespace cppa } // namespace cppa
...@@ -133,7 +133,7 @@ auto actor_namespace::proxies(node_id& node) -> proxy_map& { ...@@ -133,7 +133,7 @@ auto actor_namespace::proxies(node_id& node) -> proxy_map& {
} }
void actor_namespace::erase(node_id& inf) { void actor_namespace::erase(node_id& inf) {
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string)); CPPA_LOG_TRACE(CPPA_TARG(inf, to_string));
m_proxies.erase(inf); m_proxies.erase(inf);
} }
......
...@@ -57,7 +57,7 @@ actor_registry::value_type actor_registry::get_entry(actor_id key) const { ...@@ -57,7 +57,7 @@ actor_registry::value_type actor_registry::get_entry(actor_id key) const {
if (i != m_entries.end()) { if (i != m_entries.end()) {
return i->second; return i->second;
} }
CPPA_LOGMF(CPPA_DEBUG, self, "key not found: " << key); CPPA_LOG_DEBUG("key not found: " << key);
return {nullptr, exit_reason::not_exited}; return {nullptr, exit_reason::not_exited};
} }
...@@ -75,7 +75,7 @@ void actor_registry::put(actor_id key, const abstract_actor_ptr& value) { ...@@ -75,7 +75,7 @@ void actor_registry::put(actor_id key, const abstract_actor_ptr& value) {
} }
} }
if (add_attachable) { if (add_attachable) {
CPPA_LOGMF(CPPA_INFO, self, "added actor with ID " << key); CPPA_LOG_INFO("added actor with ID " << key);
struct eraser : attachable { struct eraser : attachable {
actor_id m_id; actor_id m_id;
actor_registry* m_registry; actor_registry* m_registry;
...@@ -96,8 +96,7 @@ void actor_registry::erase(actor_id key, std::uint32_t reason) { ...@@ -96,8 +96,7 @@ void actor_registry::erase(actor_id key, std::uint32_t reason) {
auto i = m_entries.find(key); auto i = m_entries.find(key);
if (i != m_entries.end()) { if (i != m_entries.end()) {
auto& entry = i->second; auto& entry = i->second;
CPPA_LOGMF(CPPA_INFO, self, "erased actor with ID " << key CPPA_LOG_INFO("erased actor with ID " << key << ", reason " << reason);
<< ", reason " << reason);
entry.first = nullptr; entry.first = nullptr;
entry.second = reason; entry.second = reason;
} }
...@@ -133,7 +132,7 @@ void actor_registry::dec_running() { ...@@ -133,7 +132,7 @@ void actor_registry::dec_running() {
} }
void actor_registry::await_running_count_equal(size_t expected) { void actor_registry::await_running_count_equal(size_t expected) {
CPPA_LOGMF(CPPA_TRACE, self.unchecked(), CPPA_ARG(expected)); CPPA_LOG_TRACE(CPPA_ARG(expected));
std::unique_lock<std::mutex> guard{m_running_mtx}; std::unique_lock<std::mutex> guard{m_running_mtx};
while (m_running != expected) { while (m_running != expected) {
CPPA_LOG_DEBUG("count = " << m_running.load()); CPPA_LOG_DEBUG("count = " << m_running.load());
......
...@@ -63,7 +63,7 @@ pointer advanced(pointer ptr, size_t num_bytes) { ...@@ -63,7 +63,7 @@ pointer advanced(pointer ptr, size_t num_bytes) {
inline void range_check(pointer begin, pointer end, size_t read_size) { inline void range_check(pointer begin, pointer end, size_t read_size) {
if (advanced(begin, read_size) > end) { if (advanced(begin, read_size) > end) {
CPPA_LOGF(CPPA_ERROR, self, "range_check failed"); CPPA_LOGF(CPPA_ERROR, "range_check failed");
throw out_of_range("binary_deserializer::read_range()"); throw out_of_range("binary_deserializer::read_range()");
} }
} }
...@@ -200,7 +200,7 @@ const uniform_type_info* binary_deserializer::begin_object() { ...@@ -200,7 +200,7 @@ const uniform_type_info* binary_deserializer::begin_object() {
void binary_deserializer::end_object() { } void binary_deserializer::end_object() { }
size_t binary_deserializer::begin_sequence() { size_t binary_deserializer::begin_sequence() {
CPPA_LOGMF(CPPA_TRACE, self, ""); CPPA_LOG_TRACE("");
static_assert(sizeof(size_t) >= sizeof(uint32_t), static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)"); "sizeof(size_t) < sizeof(uint32_t)");
uint32_t result; uint32_t result;
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp" #include "cppa/singletons.hpp"
#include "cppa/blocking_untyped_actor.hpp" #include "cppa/blocking_untyped_actor.hpp"
...@@ -35,8 +36,29 @@ ...@@ -35,8 +36,29 @@
namespace cppa { namespace cppa {
void blocking_untyped_actor::response_future::await(behavior& bhvr) {
m_self->dequeue_response(bhvr, m_mid);
}
void blocking_untyped_actor::await_all_other_actors_done() { void blocking_untyped_actor::await_all_other_actors_done() {
get_actor_registry()->await_running_count_equal(1); get_actor_registry()->await_running_count_equal(1);
} }
blocking_untyped_actor::response_future
blocking_untyped_actor::sync_send_tuple(const actor& dest, any_tuple what) {
auto nri = new_request_id();
dest.enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
blocking_untyped_actor::response_future
blocking_untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
auto nri = new_request_id();
get_scheduler()->delayed_send({address(), dest, nri}, rtime, std::move(what));
return {nri.response_id(), this};
}
} // namespace cppa } // namespace cppa
...@@ -70,7 +70,7 @@ class local_group : public group { ...@@ -70,7 +70,7 @@ class local_group : public group {
public: public:
void send_all_subscribers(const message_header& hdr, const any_tuple& msg) { void send_all_subscribers(const message_header& hdr, const any_tuple& msg) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", " CPPA_LOG_TRACE(CPPA_TARG(hdr.sender, to_string) << ", "
<< CPPA_TARG(msg, to_string)); << CPPA_TARG(msg, to_string));
shared_guard guard(m_mtx); shared_guard guard(m_mtx);
for (auto& s : m_subscribers) { for (auto& s : m_subscribers) {
...@@ -166,7 +166,7 @@ class local_broker : public untyped_actor { ...@@ -166,7 +166,7 @@ class local_broker : public untyped_actor {
on(atom("DOWN"), arg_match) >> [=](uint32_t) { on(atom("DOWN"), arg_match) >> [=](uint32_t) {
auto sender = last_sender(); auto sender = last_sender();
CPPA_LOGC_TRACE("cppa::local_broker", "init$DOWN", CPPA_LOGC_TRACE("cppa::local_broker", "init$DOWN",
CPPA_TARG(other, to_string)); CPPA_TARG(sender, to_string));
if (sender) m_acquaintances.erase(sender); if (sender) m_acquaintances.erase(sender);
}, },
others() >> [=] { others() >> [=] {
......
...@@ -34,7 +34,6 @@ ...@@ -34,7 +34,6 @@
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/detail/raw_access.hpp" #include "cppa/detail/raw_access.hpp"
...@@ -80,7 +79,10 @@ constexpr const char* s_default_debug_name = "actor"; ...@@ -80,7 +79,10 @@ constexpr const char* s_default_debug_name = "actor";
local_actor::local_actor() local_actor::local_actor()
: m_trap_exit(false) : m_trap_exit(false)
, m_dummy_node(), m_current_node(&m_dummy_node) , m_dummy_node(), m_current_node(&m_dummy_node)
, m_planned_exit_reason(exit_reason::not_exited) { } , m_planned_exit_reason(exit_reason::not_exited)
, m_state(actor_state::ready) {
m_node = node_id::get();
}
local_actor::~local_actor() { } local_actor::~local_actor() { }
...@@ -139,15 +141,6 @@ void local_actor::forward_message(const actor& dest, message_priority p) { ...@@ -139,15 +141,6 @@ void local_actor::forward_message(const actor& dest, message_priority p) {
id = message_id{}; id = message_id{};
} }
/* TODO:
response_handle local_actor::make_response_handle() {
auto n = m_current_node;
response_handle result{this, n->sender, n->mid.response_id()};
n->mid.mark_as_answered();
return result;
}
*/
void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) { void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) {
dest.enqueue({address(), dest, prio}, std::move(what)); dest.enqueue({address(), dest, prio}, std::move(what));
} }
...@@ -168,19 +161,9 @@ void local_actor::delayed_send_tuple(const channel&, const util::duration&, cppa ...@@ -168,19 +161,9 @@ void local_actor::delayed_send_tuple(const channel&, const util::duration&, cppa
} }
message_future local_actor::timed_sync_send_tuple(const util::duration& rtime, response_promise local_actor::make_response_promise() {
const actor& dest,
any_tuple what) {
}
message_future local_actor::sync_send_tuple(const actor& dest, any_tuple what) {
}
response_handle local_actor::make_response_handle() {
auto n = m_current_node; auto n = m_current_node;
response_handle result{address(), n->sender, n->mid.response_id()}; response_promise result{address(), n->sender, n->mid.response_id()};
n->mid.mark_as_answered(); n->mid.mark_as_answered();
return result; return result;
} }
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include <cstring> #include <cstring>
#include <fstream> #include <fstream>
#include <algorithm> #include <algorithm>
#include <pthread.h>
#ifndef CPPA_WINDOWS #ifndef CPPA_WINDOWS
#include <unistd.h> #include <unistd.h>
...@@ -51,6 +52,8 @@ namespace cppa { ...@@ -51,6 +52,8 @@ namespace cppa {
namespace { namespace {
__thread actor_id t_self_id;
template<size_t RawSize> template<size_t RawSize>
void replace_all(string& str, const char (&before)[RawSize], const char* after) { void replace_all(string& str, const char (&before)[RawSize], const char* after) {
// end(before) - 1 points to the null-terminator // end(before) - 1 points to the null-terminator
...@@ -61,6 +64,9 @@ void replace_all(string& str, const char (&before)[RawSize], const char* after) ...@@ -61,6 +64,9 @@ void replace_all(string& str, const char (&before)[RawSize], const char* after)
} }
} }
constexpr struct pop_aid_log_event_t { constexpr pop_aid_log_event_t() { } }
pop_aid_log_event;
struct log_event { struct log_event {
log_event* next; log_event* next;
string msg; string msg;
...@@ -72,11 +78,11 @@ class logging_impl : public logging { ...@@ -72,11 +78,11 @@ class logging_impl : public logging {
void initialize() { void initialize() {
m_thread = thread([this] { (*this)(); }); m_thread = thread([this] { (*this)(); });
log("TRACE", "logging", "run", __FILE__, __LINE__, invalid_actor_addr, "ENTRY"); log("TRACE", "logging", "run", __FILE__, __LINE__, "ENTRY");
} }
void destroy() { void destroy() {
log("TRACE", "logging", "run", __FILE__, __LINE__, invalid_actor_addr, "EXIT"); log("TRACE", "logging", "run", __FILE__, __LINE__, "EXIT");
// an empty string means: shut down // an empty string means: shut down
m_queue.push_back(new log_event{0, ""}); m_queue.push_back(new log_event{0, ""});
m_thread.join(); m_thread.join();
...@@ -98,13 +104,18 @@ class logging_impl : public logging { ...@@ -98,13 +104,18 @@ class logging_impl : public logging {
} }
} }
actor_id set_aid(actor_id aid) override {
actor_id prev = t_self_id;
t_self_id = aid;
return prev;
}
void log(const char* level, void log(const char* level,
const char* c_class_name, const char* c_class_name,
const char* function_name, const char* function_name,
const char* c_full_file_name, const char* c_full_file_name,
int line_num, int line_num,
actor_addr, const std::string& msg) override {
const std::string& msg) {
string class_name = c_class_name; string class_name = c_class_name;
replace_all(class_name, "::", "."); replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymous namespace)", "$anon$"); replace_all(class_name, "(anonymous namespace)", "$anon$");
...@@ -117,34 +128,17 @@ class logging_impl : public logging { ...@@ -117,34 +128,17 @@ class logging_impl : public logging {
else file_name = string(i, full_file_name.end()); else file_name = string(i, full_file_name.end());
} }
else file_name = move(full_file_name); 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";
}
else if (from->is_proxy()) oss << to_string(from);
else {
# ifdef CPPA_DEBUG_MODE
oss << from.downcast<local_actor>()->debug_name();
# else // CPPA_DEBUG_MODE
oss << from->id() << "@local";
# endif // CPPA_DEBUG_MODE
}
*/
return oss;
};
ostringstream line; ostringstream line;
line << time(0) << " " line << time(0) << " "
<< level << " "; << level << " "
print_from(line) << " " << "actor" << t_self_id << " "
<< this_thread::get_id() << " " << this_thread::get_id() << " "
<< class_name << " " << class_name << " "
<< function_name << " " << function_name << " "
<< file_name << ":" << line_num << " " << file_name << ":" << line_num << " "
<< msg << msg
<< endl; << endl;
m_queue.push_back(new log_event{0, line.str()}); m_queue.push_back(new log_event{nullptr, line.str()});
} }
private: private:
...@@ -160,17 +154,16 @@ logging::trace_helper::trace_helper(std::string class_name, ...@@ -160,17 +154,16 @@ logging::trace_helper::trace_helper(std::string class_name,
const char* fun_name, const char* fun_name,
const char* file_name, const char* file_name,
int line_num, int line_num,
actor_addr ptr,
const std::string& msg) const std::string& msg)
: m_class(std::move(class_name)), m_fun_name(fun_name) : 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)) { , m_file_name(file_name), m_line_num(line_num) {
get_logger()->log("TRACE", m_class.c_str(), fun_name, get_logger()->log("TRACE", m_class.c_str(), fun_name,
file_name, line_num, m_self, "ENTRY " + msg); file_name, line_num, "ENTRY " + msg);
} }
logging::trace_helper::~trace_helper() { logging::trace_helper::~trace_helper() {
get_logger()->log("TRACE", m_class.c_str(), m_fun_name, get_logger()->log("TRACE", m_class.c_str(), m_fun_name,
m_file_name, m_line_num, m_self, "EXIT"); m_file_name, m_line_num, "EXIT");
} }
logging::~logging() { } logging::~logging() { }
......
...@@ -169,13 +169,13 @@ class middleman_impl : public middleman { ...@@ -169,13 +169,13 @@ class middleman_impl : public middleman {
} }
peer* get_peer(const node_id& node) override { peer* get_peer(const node_id& node) override {
CPPA_LOG_TRACE("n = " << to_string(n)); CPPA_LOG_TRACE(CPPA_TARG(node, to_string));
auto i = m_peers.find(node); auto i = m_peers.find(node);
if (i != m_peers.end()) { if (i != m_peers.end()) {
CPPA_LOG_DEBUG("result = " << i->second.impl); CPPA_LOG_DEBUG("result = " << i->second.impl);
return i->second.impl; return i->second.impl;
} }
CPPA_LOGMF(CPPA_DEBUG, self, "result = nullptr"); CPPA_LOG_DEBUG("result = nullptr");
return nullptr; return nullptr;
} }
...@@ -310,21 +310,21 @@ class middleman_overseer : public continuable { ...@@ -310,21 +310,21 @@ class middleman_overseer : public continuable {
static constexpr size_t num_dummies = 64; static constexpr size_t num_dummies = 64;
uint8_t dummies[num_dummies]; uint8_t dummies[num_dummies];
auto read_result = ::read(read_handle(), dummies, num_dummies); auto read_result = ::read(read_handle(), dummies, num_dummies);
CPPA_LOGMF(CPPA_DEBUG, self, "read " << read_result << " messages from queue"); CPPA_LOG_DEBUG("read " << read_result << " messages from queue");
if (read_result < 0) { if (read_result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
// try again later // try again later
return read_continue_later; return read_continue_later;
} }
else { else {
CPPA_LOGMF(CPPA_ERROR, self, "cannot read from pipe"); CPPA_LOG_ERROR("cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe"); CPPA_CRITICAL("cannot read from pipe");
} }
} }
for (int i = 0; i < read_result; ++i) { for (int i = 0; i < read_result; ++i) {
unique_ptr<middleman_event> msg(m_queue.try_pop()); unique_ptr<middleman_event> msg(m_queue.try_pop());
if (!msg) { if (!msg) {
CPPA_LOGMF(CPPA_ERROR, self, "nullptr dequeued"); CPPA_LOG_ERROR("nullptr dequeued");
CPPA_CRITICAL("nullptr dequeued"); CPPA_CRITICAL("nullptr dequeued");
} }
CPPA_LOGF_DEBUG("execute run_later functor"); CPPA_LOGF_DEBUG("execute run_later functor");
......
...@@ -116,7 +116,7 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -116,7 +116,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
tmp.events = to_poll_bitmask(new_bitmask); tmp.events = to_poll_bitmask(new_bitmask);
tmp.revents = 0; tmp.revents = 0;
m_pollset.insert(iter, tmp); m_pollset.insert(iter, tmp);
CPPA_LOGMF(CPPA_DEBUG, self, "inserted new element"); CPPA_LOG_DEBUG("inserted new element");
break; break;
} }
case fd_meta_event::erase: { case fd_meta_event::erase: {
...@@ -124,7 +124,7 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -124,7 +124,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
"m_meta and m_pollset out of sync; " "m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)"); "no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) { if (iter != last && iter->fd == fd) {
CPPA_LOGMF(CPPA_DEBUG, self, "erased element"); CPPA_LOG_DEBUG("erased element");
m_pollset.erase(iter); m_pollset.erase(iter);
} }
break; break;
...@@ -134,7 +134,7 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -134,7 +134,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
"m_meta and m_pollset out of sync; " "m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)"); "no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) { if (iter != last && iter->fd == fd) {
CPPA_LOGMF(CPPA_DEBUG, self, "updated bitmask"); CPPA_LOG_DEBUG("updated bitmask");
iter->events = to_poll_bitmask(new_bitmask); iter->events = to_poll_bitmask(new_bitmask);
} }
break; break;
......
...@@ -143,7 +143,7 @@ continue_reading_result peer::continue_reading() { ...@@ -143,7 +143,7 @@ continue_reading_result peer::continue_reading() {
m_meta_msg->deserialize(&msg, &bd); m_meta_msg->deserialize(&msg, &bd);
} }
catch (exception& e) { catch (exception& e) {
CPPA_LOGMF(CPPA_ERROR, self, "exception during read_message: " CPPA_LOG_ERROR("exception during read_message: "
<< detail::demangle(typeid(e)) << detail::demangle(typeid(e))
<< ", what(): " << e.what()); << ", what(): " << e.what());
return read_failure; return read_failure;
...@@ -192,22 +192,22 @@ void peer::monitor(const actor_addr&, ...@@ -192,22 +192,22 @@ void peer::monitor(const actor_addr&,
actor_id aid) { actor_id aid) {
CPPA_LOG_TRACE(CPPA_MARG(node, get) << ", " << CPPA_ARG(aid)); CPPA_LOG_TRACE(CPPA_MARG(node, get) << ", " << CPPA_ARG(aid));
if (!node) { if (!node) {
CPPA_LOGMF(CPPA_ERROR, self, "received MONITOR from invalid peer"); CPPA_LOG_ERROR("received MONITOR from invalid peer");
return; return;
} }
auto entry = get_actor_registry()->get_entry(aid); auto entry = get_actor_registry()->get_entry(aid);
auto pself = node_id::get(); auto pself = node_id::get();
if (*node == *pself) { if (*node == *pself) {
CPPA_LOGMF(CPPA_ERROR, self, "received 'MONITOR' from pself"); CPPA_LOG_ERROR("received 'MONITOR' from pself");
} }
else if (entry.first == nullptr) { else if (entry.first == nullptr) {
if (entry.second == exit_reason::not_exited) { if (entry.second == exit_reason::not_exited) {
CPPA_LOGMF(CPPA_ERROR, self, "received MONITOR for unknown " CPPA_LOG_ERROR("received MONITOR for unknown "
"actor id: " << aid); "actor id: " << aid);
} }
else { else {
CPPA_LOGMF(CPPA_DEBUG, self, "received MONITOR for an actor " CPPA_LOG_DEBUG("received MONITOR for an actor "
"that already finished " "that already finished "
"execution; reply KILL_PROXY"); "execution; reply KILL_PROXY");
// this actor already finished execution; // this actor already finished execution;
...@@ -217,7 +217,7 @@ void peer::monitor(const actor_addr&, ...@@ -217,7 +217,7 @@ void peer::monitor(const actor_addr&,
} }
} }
else { else {
CPPA_LOGMF(CPPA_DEBUG, self, "attach functor to " << entry.first.get()); CPPA_LOG_DEBUG("attach functor to " << entry.first.get());
auto mm = parent(); auto mm = parent();
entry.first->address()->attach_functor([=](uint32_t reason) { entry.first->address()->attach_functor([=](uint32_t reason) {
mm->run_later([=] { mm->run_later([=] {
...@@ -240,11 +240,11 @@ void peer::kill_proxy(const actor_addr& sender, ...@@ -240,11 +240,11 @@ void peer::kill_proxy(const actor_addr& sender,
<< ", " << CPPA_ARG(aid) << ", " << CPPA_ARG(aid)
<< ", " << CPPA_ARG(reason)); << ", " << CPPA_ARG(reason));
if (!node) { if (!node) {
CPPA_LOGMF(CPPA_ERROR, self, "node = nullptr"); CPPA_LOG_ERROR("node = nullptr");
return; return;
} }
if (sender != nullptr) { if (sender != nullptr) {
CPPA_LOGMF(CPPA_ERROR, self, "sender != nullptr"); CPPA_LOG_ERROR("sender != nullptr");
return; return;
} }
auto proxy = parent()->get_namespace().get(*node, aid); auto proxy = parent()->get_namespace().get(*node, aid);
...@@ -273,8 +273,8 @@ void peer::link(const actor_addr& sender, const actor_addr& receiver) { ...@@ -273,8 +273,8 @@ void peer::link(const actor_addr& sender, const actor_addr& receiver) {
// this message is sent from default_actor_proxy in link_to and // this message is sent from default_actor_proxy in link_to and
// establish_backling to cause the original actor (sender) to establish // establish_backling to cause the original actor (sender) to establish
// a link to ptr as well // a link to ptr as well
CPPA_LOG_TRACE(CPPA_MARG(sender, get) CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< ", " << CPPA_MARG(ptr, get)); << CPPA_TARG(receiver, to_string));
CPPA_LOG_ERROR_IF(!sender, "received 'LINK' from invalid sender"); CPPA_LOG_ERROR_IF(!sender, "received 'LINK' from invalid sender");
CPPA_LOG_ERROR_IF(!receiver, "received 'LINK' with invalid receiver"); CPPA_LOG_ERROR_IF(!receiver, "received 'LINK' with invalid receiver");
if (!sender || !receiver) return; if (!sender || !receiver) return;
...@@ -297,8 +297,8 @@ void peer::link(const actor_addr& sender, const actor_addr& receiver) { ...@@ -297,8 +297,8 @@ void peer::link(const actor_addr& sender, const actor_addr& receiver) {
} }
void peer::unlink(const actor_addr& sender, const actor_addr& receiver) { void peer::unlink(const actor_addr& sender, const actor_addr& receiver) {
CPPA_LOG_TRACE(CPPA_MARG(sender, get) CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< ", " << CPPA_MARG(ptr, get)); << CPPA_TARG(receiver, to_string));
CPPA_LOG_ERROR_IF(!sender, "received 'UNLINK' from invalid sender"); CPPA_LOG_ERROR_IF(!sender, "received 'UNLINK' from invalid sender");
CPPA_LOG_ERROR_IF(!receiver, "received 'UNLINK' with invalid target"); CPPA_LOG_ERROR_IF(!receiver, "received 'UNLINK' with invalid target");
if (!sender || !receiver) return; if (!sender || !receiver) return;
...@@ -357,13 +357,13 @@ void peer::enqueue_impl(const message_header& hdr, const any_tuple& msg) { ...@@ -357,13 +357,13 @@ void peer::enqueue_impl(const message_header& hdr, const any_tuple& msg) {
wbuf.write(sizeof(uint32_t), &size); wbuf.write(sizeof(uint32_t), &size);
try { bs << hdr << msg; } try { bs << hdr << msg; }
catch (exception& e) { catch (exception& e) {
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e)); CPPA_LOG_ERROR(to_verbose_string(e));
cerr << "*** exception in peer::enqueue; " cerr << "*** exception in peer::enqueue; "
<< to_verbose_string(e) << to_verbose_string(e)
<< endl; << endl;
return; return;
} }
CPPA_LOGMF(CPPA_DEBUG, self, "serialized: " << to_string(hdr) << " " << to_string(msg)); CPPA_LOG_DEBUG("serialized: " << to_string(hdr) << " " << to_string(msg));
size = (wbuf.size() - before) - sizeof(std::uint32_t); size = (wbuf.size() - before) - sizeof(std::uint32_t);
// update size in buffer // update size in buffer
memcpy(wbuf.offset_data(before), &size, sizeof(std::uint32_t)); memcpy(wbuf.offset_data(before), &size, sizeof(std::uint32_t));
......
...@@ -69,8 +69,7 @@ remote_actor_proxy::~remote_actor_proxy() { ...@@ -69,8 +69,7 @@ remote_actor_proxy::~remote_actor_proxy() {
mm->run_later([aid, node, mm] { mm->run_later([aid, node, mm] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy", CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"~remote_actor_proxy$run_later", "~remote_actor_proxy$run_later",
"node = " << to_string(*node) << ", aid " << aid "node = " << to_string(*node) << ", aid " << aid);
<< ", proto = " << to_string(proto->identifier()));
mm->get_namespace().erase(*node, aid); mm->get_namespace().erase(*node, aid);
auto p = mm->get_peer(*node); auto p = mm->get_peer(*node);
if (p && p->erase_on_last_proxy_exited()) { if (p && p->erase_on_last_proxy_exited()) {
......
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
#include <utility> #include <utility>
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/response_handle.hpp" #include "cppa/response_promise.hpp"
#include "cppa/detail/raw_access.hpp" #include "cppa/detail/raw_access.hpp"
...@@ -39,27 +39,22 @@ using std::move; ...@@ -39,27 +39,22 @@ using std::move;
namespace cppa { namespace cppa {
response_handle::response_handle(const actor_addr& from, response_promise::response_promise(const actor_addr& from,
const actor_addr& to, const actor_addr& to,
const message_id& id) const message_id& id)
: m_from(from), m_to(to), m_id(id) { : m_from(from), m_to(to), m_id(id) {
CPPA_REQUIRE(id.is_response() || !id.valid()); CPPA_REQUIRE(id.is_response() || !id.valid());
} }
bool response_handle::valid() const { response_promise::operator bool() const {
return m_to != nullptr; return m_to != nullptr;
} }
bool response_handle::synchronous() const { void response_promise::deliver(any_tuple msg) {
return m_id.valid(); if (m_to) {
}
void response_handle::apply(any_tuple msg) const {
std::cout << "response_handle::apply\n";
if (valid()) {
auto ptr = detail::actor_addr_cast<abstract_actor>(m_to); auto ptr = detail::actor_addr_cast<abstract_actor>(m_to);
ptr->enqueue({m_from, ptr, m_id}, move(msg)); ptr->enqueue({m_from, ptr, m_id}, move(msg));
std::cout << "response_handle::apply: after ptr->enqueue\n"; m_to = invalid_actor_addr;
} }
} }
......
...@@ -29,9 +29,11 @@ ...@@ -29,9 +29,11 @@
#include "cppa/policy.hpp" #include "cppa/policy.hpp"
#include "cppa/singletons.hpp"
#include "cppa/scoped_actor.hpp" #include "cppa/scoped_actor.hpp"
#include "cppa/detail/proper_actor.hpp" #include "cppa/detail/proper_actor.hpp"
#include "cppa/detail/actor_registry.hpp"
namespace cppa { namespace cppa {
...@@ -52,8 +54,14 @@ blocking_untyped_actor* alloc() { ...@@ -52,8 +54,14 @@ blocking_untyped_actor* alloc() {
} // namespace <anonymous> } // namespace <anonymous>
scoped_actor::scoped_actor() : m_self(alloc()) { } scoped_actor::scoped_actor() : m_self(alloc()) {
m_prev = CPPA_SET_AID(m_self->id());
get_actor_registry()->inc_running();
}
scoped_actor::~scoped_actor() { } scoped_actor::~scoped_actor() {
get_actor_registry()->dec_running();
CPPA_SET_AID(m_prev);
}
} // namespace cppa } // namespace cppa
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
namespace cppa { namespace cppa {
message_future sync_send_tuple(actor_destination dest, any_tuple what) { response_future sync_send_tuple(actor_destination dest, any_tuple what) {
if (!dest.receiver) throw std::invalid_argument("whom == nullptr"); if (!dest.receiver) throw std::invalid_argument("whom == nullptr");
auto req = self->new_request_id(); auto req = self->new_request_id();
message_header hdr{self, std::move(dest.receiver), req, dest.priority}; message_header hdr{self, std::move(dest.receiver), req, dest.priority};
...@@ -55,7 +55,7 @@ void delayed_send_tuple(channel_destination dest, ...@@ -55,7 +55,7 @@ void delayed_send_tuple(channel_destination dest,
} }
} }
message_future timed_sync_send_tuple(actor_destination dest, response_future timed_sync_send_tuple(actor_destination dest,
const util::duration& rtime, const util::duration& rtime,
any_tuple what) { any_tuple what) {
auto mf = sync_send_tuple(std::move(dest), std::move(what)); auto mf = sync_send_tuple(std::move(dest), std::move(what));
......
...@@ -74,21 +74,21 @@ std::atomic<logging*> s_logger; ...@@ -74,21 +74,21 @@ std::atomic<logging*> s_logger;
} // namespace <anonymous> } // namespace <anonymous>
void singleton_manager::shutdown() { void singleton_manager::shutdown() {
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown scheduler"); CPPA_LOGF_DEBUG("shutdown scheduler");
destroy(s_scheduler); destroy(s_scheduler);
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown middleman"); CPPA_LOGF_DEBUG("shutdown middleman");
destroy(s_middleman); destroy(s_middleman);
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
// it's safe to delete all other singletons now // it's safe to delete all other singletons now
CPPA_LOGF(CPPA_DEBUG, nullptr, "close OpenCL metainfo"); CPPA_LOGF_DEBUG("close OpenCL metainfo");
destroy(s_opencl_metainfo); destroy(s_opencl_metainfo);
CPPA_LOGF(CPPA_DEBUG, nullptr, "close actor registry"); CPPA_LOGF_DEBUG("close actor registry");
destroy(s_actor_registry); destroy(s_actor_registry);
CPPA_LOGF(CPPA_DEBUG, nullptr, "shutdown group manager"); CPPA_LOGF_DEBUG("shutdown group manager");
destroy(s_group_manager); destroy(s_group_manager);
CPPA_LOGF(CPPA_DEBUG, nullptr, "destroy empty tuple singleton"); CPPA_LOGF_DEBUG("destroy empty tuple singleton");
destroy(s_empty_tuple); destroy(s_empty_tuple);
CPPA_LOGF(CPPA_DEBUG, nullptr, "clear type info map"); CPPA_LOGF_DEBUG("clear type info map");
destroy(s_uniform_type_info_map); destroy(s_uniform_type_info_map);
destroy(s_logger); destroy(s_logger);
} }
......
...@@ -101,16 +101,15 @@ struct thread_pool_scheduler::worker { ...@@ -101,16 +101,15 @@ struct thread_pool_scheduler::worker {
job_ptr job = nullptr; job_ptr job = nullptr;
for (;;) { for (;;) {
aggressive(job) || moderate(job) || relaxed(job); aggressive(job) || moderate(job) || relaxed(job);
CPPA_LOGMF(CPPA_DEBUG, self, "dequeued new job"); CPPA_LOG_DEBUG("dequeued new job");
if (job == m_dummy) { if (job == m_dummy) {
CPPA_LOGMF(CPPA_DEBUG, self, "received dummy (quit)"); CPPA_LOG_DEBUG("received dummy (quit)");
// dummy of doom received ... // dummy of doom received ...
m_job_queue->push_back(job); // kill the next guy m_job_queue->push_back(job); // kill the next guy
return; // and say goodbye return; // and say goodbye
} }
CPPA_LOG_DEBUG("resume actor with ID " << job->id());
if (job->resume(&fself) == resumable::done) { if (job->resume(&fself) == resumable::done) {
CPPA_LOGMF(CPPA_DEBUG, self, "actor is done"); CPPA_LOG_DEBUG("actor is done");
/*FIXME bool hidden = job->is_hidden(); /*FIXME bool hidden = job->is_hidden();
job->deref(); job->deref();
if (!hidden)*/ get_actor_registry()->dec_running(); if (!hidden)*/ get_actor_registry()->dec_running();
...@@ -156,11 +155,11 @@ void thread_pool_scheduler::initialize() { ...@@ -156,11 +155,11 @@ void thread_pool_scheduler::initialize() {
void thread_pool_scheduler::destroy() { void thread_pool_scheduler::destroy() {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
m_queue.push_back(&m_dummy); m_queue.push_back(&m_dummy);
CPPA_LOGMF(CPPA_DEBUG, self, "join supervisor"); CPPA_LOG_DEBUG("join supervisor");
m_supervisor.join(); m_supervisor.join();
// make sure job queue is empty, because destructor of m_queue would // make sure job queue is empty, because destructor of m_queue would
// otherwise delete elements it shouldn't // otherwise delete elements it shouldn't
CPPA_LOGMF(CPPA_DEBUG, self, "flush queue"); CPPA_LOG_DEBUG("flush queue");
auto ptr = m_queue.try_pop(); auto ptr = m_queue.try_pop();
while (ptr != nullptr) { while (ptr != nullptr) {
if (ptr != &m_dummy) { if (ptr != &m_dummy) {
......
...@@ -66,8 +66,7 @@ using namespace detail; ...@@ -66,8 +66,7 @@ using namespace detail;
using namespace io; using namespace io;
void publish(actor whom, std::unique_ptr<acceptor> aptr) { void publish(actor whom, std::unique_ptr<acceptor> aptr) {
CPPA_LOG_TRACE(CPPA_TARG(whom, to_string) << ", " << CPPA_MARG(ptr, get) CPPA_LOGF_TRACE(CPPA_TARG(whom, to_string) << ", " << CPPA_MARG(aptr, get));
<< ", args.size() = " << args.size());
if (!whom) return; if (!whom) return;
get_actor_registry()->put(whom->id(), detail::actor_addr_cast<abstract_actor>(whom)); get_actor_registry()->put(whom->id(), detail::actor_addr_cast<abstract_actor>(whom));
auto mm = get_middleman(); auto mm = get_middleman();
...@@ -80,7 +79,7 @@ void publish(actor whom, std::uint16_t port, const char* addr) { ...@@ -80,7 +79,7 @@ void publish(actor whom, std::uint16_t port, const char* addr) {
} }
actor remote_actor(stream_ptr_pair io) { actor remote_actor(stream_ptr_pair io) {
CPPA_LOG_TRACE("io{" << io.first.get() << ", " << io.second.get() << "}"); CPPA_LOGF_TRACE("io{" << io.first.get() << ", " << io.second.get() << "}");
auto pinf = node_id::get(); auto pinf = node_id::get();
std::uint32_t process_id = pinf->process_id(); std::uint32_t process_id = pinf->process_id();
// throws on error // throws on error
...@@ -111,7 +110,7 @@ actor remote_actor(stream_ptr_pair io) { ...@@ -111,7 +110,7 @@ actor remote_actor(stream_ptr_pair io) {
q.push_back(new remote_actor_result{0, res}); q.push_back(new remote_actor_result{0, res});
}); });
std::unique_ptr<remote_actor_result> result(q.pop()); std::unique_ptr<remote_actor_result> result(q.pop());
CPPA_LOGF_DEBUG("result = " << result->value.get()); CPPA_LOGF_DEBUG(CPPA_MARG(result, get));
return result->value; return result->value;
} }
......
...@@ -59,8 +59,6 @@ ...@@ -59,8 +59,6 @@
#include "cppa/detail/uniform_type_info_map.hpp" #include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp" #include "cppa/detail/default_uniform_type_info_impl.hpp"
#include "cppa/message_header.hpp"
namespace cppa { namespace cppa {
namespace { inline detail::uniform_type_info_map& uti_map() { namespace { inline detail::uniform_type_info_map& uti_map() {
......
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include <algorithm> #include <algorithm>
#include <type_traits> #include <type_traits>
#include "cppa/group.hpp"
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/announce.hpp" #include "cppa/announce.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
...@@ -296,12 +297,16 @@ void serialize_impl(const message_header& hdr, serializer* sink) { ...@@ -296,12 +297,16 @@ void serialize_impl(const message_header& hdr, serializer* sink) {
serialize_impl(hdr.sender, sink); serialize_impl(hdr.sender, sink);
serialize_impl(hdr.receiver, sink); serialize_impl(hdr.receiver, sink);
sink->write_value(hdr.id.integer_value()); sink->write_value(hdr.id.integer_value());
sink->write_value(static_cast<uint32_t>(hdr.priority));
} }
void deserialize_impl(message_header& hdr, deserializer* source) { void deserialize_impl(message_header& hdr, deserializer* source) {
deserialize_impl(hdr.sender, source); deserialize_impl(hdr.sender, source);
deserialize_impl(hdr.receiver, source); deserialize_impl(hdr.receiver, source);
hdr.id = message_id::from_integer_value(source->read<std::uint64_t>()); hdr.id = message_id::from_integer_value(source->read<std::uint64_t>());
auto prio = source->read<std::uint32_t>();
//TODO: check whether integer is actually a valid priority
hdr.priority = static_cast<message_priority>(prio);
} }
void serialize_impl(const node_id_ptr& ptr, serializer* sink) { void serialize_impl(const node_id_ptr& ptr, serializer* sink) {
......
...@@ -28,12 +28,44 @@ ...@@ -28,12 +28,44 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp"
#include "cppa/message_id.hpp"
#include "cppa/untyped_actor.hpp" #include "cppa/untyped_actor.hpp"
namespace cppa { namespace cppa {
continue_helper& continue_helper::continue_with(behavior::continuation_fun fun) {
auto ref_opt = m_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("failed to add continuation");
return *this;
}
void untyped_actor::forward_to(const actor&) { void untyped_actor::forward_to(const actor&) {
} }
untyped_actor::response_future untyped_actor::sync_send_tuple(const actor& dest,
any_tuple what) {
auto nri = new_request_id();
dest.enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
untyped_actor::response_future
untyped_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
auto nri = new_request_id();
get_scheduler()->delayed_send({address(), dest, nri}, rtime, std::move(what));
return {nri.response_id(), this};
}
} // namespace cppa } // namespace cppa
...@@ -34,7 +34,7 @@ void cppa_unexpected_timeout(const char* fname, size_t line_num); ...@@ -34,7 +34,7 @@ void cppa_unexpected_timeout(const char* fname, size_t line_num);
#define CPPA_PRINT(message) CPPA_PRINTC(__FILE__, __LINE__, message) #define CPPA_PRINT(message) CPPA_PRINTC(__FILE__, __LINE__, message)
#define CPPA_PRINTERRC(fname, linenum, msg) \ #define CPPA_PRINTERRC(fname, linenum, msg) \
CPPA_LOGF(CPPA_ERROR, ::cppa::self, CPPA_STREAMIFY(fname, linenum, msg)); \ CPPA_LOGF(CPPA_ERROR, CPPA_STREAMIFY(fname, linenum, msg)); \
std::cerr << "ERROR: " << CPPA_STREAMIFY(fname, linenum, msg) \ std::cerr << "ERROR: " << CPPA_STREAMIFY(fname, linenum, msg) \
<< std::endl << std::endl
......
...@@ -330,32 +330,36 @@ struct slave : untyped_actor { ...@@ -330,32 +330,36 @@ struct slave : untyped_actor {
void test_serial_reply() { void test_serial_reply() {
auto mirror_behavior = [=](untyped_actor* self, int num) { auto mirror_behavior = [=](untyped_actor* self, int num) {
self->become(others() >> [=]() -> any_tuple { self->become(others() >> [=]() -> any_tuple {
cout << "right back at you from " << num CPPA_LOGF_INFO("return self->last_dequeued()");
<< "; ID: " << self->id() << endl;
return self->last_dequeued(); return self->last_dequeued();
}); });
}; };
auto master = spawn([=](untyped_actor* self) { auto master = spawn([=](untyped_actor* self) {
cout << "ID of master: " << self->id() << endl; cout << "ID of master: " << self->id() << endl;
// produce 5 mirror actors // spawn 5 mirror actors
auto c0 = self->spawn<linked>(mirror_behavior, 0); auto c0 = self->spawn<linked>(mirror_behavior, 0);
auto c1 = self->spawn<linked>(mirror_behavior, 1); auto c1 = self->spawn<linked>(mirror_behavior, 1);
auto c2 = self->spawn<linked>(mirror_behavior, 2); auto c2 = self->spawn<linked>(mirror_behavior, 2);
auto c3 = self->spawn<linked>(mirror_behavior, 3); auto c3 = self->spawn<linked>(mirror_behavior, 3);
auto c4 = self->spawn<linked>(mirror_behavior, 4); auto c4 = self->spawn<linked>(mirror_behavior, 4);
self->become ( self->become (
on(atom("hi there")) >> [=] { on(atom("hi there")) >> [=]() -> continue_helper {
// * CPPA_LOGF_INFO("received 'hi there'");
return self->sync_send(c0, atom("sub0")).then( return self->sync_send(c0, atom("sub0")).then(
on(atom("sub0")) >> [=] { on(atom("sub0")) >> [=]() -> continue_helper {
CPPA_LOGF_INFO("received 'sub0'");
return self->sync_send(c1, atom("sub1")).then( return self->sync_send(c1, atom("sub1")).then(
on(atom("sub1")) >> [=] { on(atom("sub1")) >> [=]() -> continue_helper {
CPPA_LOGF_INFO("received 'sub1'");
return self->sync_send(c2, atom("sub2")).then( return self->sync_send(c2, atom("sub2")).then(
on(atom("sub2")) >> [=] { on(atom("sub2")) >> [=]() -> continue_helper {
CPPA_LOGF_INFO("received 'sub2'");
return self->sync_send(c3, atom("sub3")).then( return self->sync_send(c3, atom("sub3")).then(
on(atom("sub3")) >> [=] { on(atom("sub3")) >> [=]() -> continue_helper {
CPPA_LOGF_INFO("received 'sub3'");
return self->sync_send(c4, atom("sub4")).then( return self->sync_send(c4, atom("sub4")).then(
on(atom("sub4")) >> [=] { on(atom("sub4")) >> [=]() -> atom_value {
CPPA_LOGF_INFO("received 'sub4'");
return atom("hiho"); return atom("hiho");
} }
); );
...@@ -455,31 +459,34 @@ void test_continuation() { ...@@ -455,31 +459,34 @@ void test_continuation() {
await_all_actors_done(); await_all_actors_done();
} }
void test_spawn() { void test_simple_reply_response() {
scoped_actor self; scoped_actor self;
auto s = spawn([](untyped_actor* self) -> behavior { auto s = spawn([](untyped_actor* self) -> behavior {
return ( return (
others() >> [=]() -> any_tuple { others() >> [=]() -> any_tuple {
cout << "received: " << to_string(self->last_dequeued()) << endl; CPPA_CHECK(self->last_dequeued() == make_any_tuple(atom("hello")));
self->quit();
return self->last_dequeued(); return self->last_dequeued();
} }
); );
}); });
cout << "spawned actor, waiting for response" << endl;
self->send(s, atom("hello")); self->send(s, atom("hello"));
self->receive( self->receive(
others() >> [&] { others() >> [&] {
cout << "received: " << to_string(self->last_dequeued()) << endl; CPPA_CHECK(self->last_dequeued() == make_any_tuple(atom("hello")));
} }
); );
self->await_all_other_actors_done(); self->await_all_other_actors_done();
return; }
void test_spawn() {
test_simple_reply_response();
test_serial_reply(); test_serial_reply();
return;
test_or_else(); test_or_else();
test_continuation(); test_continuation();
scoped_actor self;
// check whether detached actors and scheduled actors interact w/o errors // check whether detached actors and scheduled actors interact w/o errors
auto m = spawn<master, detached>(); auto m = spawn<master, detached>();
spawn<slave>(m); spawn<slave>(m);
......
...@@ -105,7 +105,7 @@ struct D : popular_actor { ...@@ -105,7 +105,7 @@ struct D : popular_actor {
return ( return (
others() >> [=] { others() >> [=] {
/* /*
response_handle handle = make_response_handle(); response_promise handle = make_response_promise();
sync_send_tuple(buddy(), last_dequeued()).then([=] { sync_send_tuple(buddy(), last_dequeued()).then([=] {
reply_to(handle, last_dequeued()); reply_to(handle, last_dequeued());
quit(); quit();
......
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