Commit 7ede617a authored by neverlord's avatar neverlord

timed receive

parent 22541dc5
...@@ -141,6 +141,7 @@ nobase_library_include_HEADERS = \ ...@@ -141,6 +141,7 @@ nobase_library_include_HEADERS = \
cppa/primitive_type.hpp \ cppa/primitive_type.hpp \
cppa/primitive_variant.hpp \ cppa/primitive_variant.hpp \
cppa/process_information.hpp \ cppa/process_information.hpp \
cppa/receive.hpp \
cppa/ref_counted.hpp \ cppa/ref_counted.hpp \
cppa/scheduler.hpp \ cppa/scheduler.hpp \
cppa/scheduling_hint.hpp \ cppa/scheduling_hint.hpp \
......
...@@ -214,3 +214,4 @@ src/invoke_rules.cpp ...@@ -214,3 +214,4 @@ src/invoke_rules.cpp
src/abstract_tuple.cpp src/abstract_tuple.cpp
cppa/util/duration.hpp cppa/util/duration.hpp
src/duration.cpp src/duration.cpp
cppa/receive.hpp
...@@ -154,12 +154,20 @@ inline void quit(std::uint32_t reason) ...@@ -154,12 +154,20 @@ inline void quit(std::uint32_t reason)
*/ */
void receive_loop(invoke_rules& rules); void receive_loop(invoke_rules& rules);
void receive_loop(timed_invoke_rules& rules);
inline void receive_loop(invoke_rules&& rules) inline void receive_loop(invoke_rules&& rules)
{ {
invoke_rules tmp(std::move(rules)); invoke_rules tmp(std::move(rules));
receive_loop(tmp); receive_loop(tmp);
} }
inline void receive_loop(timed_invoke_rules&& rules)
{
timed_invoke_rules tmp(std::move(rules));
receive_loop(tmp);
}
template<typename Head, typename... Tail> template<typename Head, typename... Tail>
void receive_loop(invoke_rules&& rules, Head&& head, Tail&&... tail) void receive_loop(invoke_rules&& rules, Head&& head, Tail&&... tail)
{ {
...@@ -208,63 +216,7 @@ detail::do_receive_helper do_receive(Args&&... args) ...@@ -208,63 +216,7 @@ detail::do_receive_helper do_receive(Args&&... args)
return detail::do_receive_helper(std::forward<Args>(args)...); return detail::do_receive_helper(std::forward<Args>(args)...);
} }
/**
* @brief Dequeues the next message from the mailbox.
*/
inline const message& receive()
{
return self()->mailbox().dequeue();
}
/**
* @brief Dequeues the next message from the mailbox that's matched
* by @p rules and executes the corresponding callback.
*/
inline void receive(invoke_rules& rules)
{
self()->mailbox().dequeue(rules);
}
inline void receive(invoke_rules&& rules)
{
invoke_rules tmp(std::move(rules));
self()->mailbox().dequeue(tmp);
}
template<typename Head, typename... Tail>
void receive(invoke_rules&& rules, Head&& head, Tail&&... tail)
{
invoke_rules tmp(std::move(rules));
receive(tmp.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
template<typename Head, typename... Tail>
void receive(invoke_rules& rules, Head&& head, Tail&&... tail)
{
receive(rules.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
/**
* @brief Tries to dequeue the next message from the mailbox.
* @return @p true if a messages was dequeued;
* @p false if the mailbox is empty
*/
inline bool try_receive(message& msg)
{
return self()->mailbox().try_dequeue(msg);
}
/**
* @brief Tries to dequeue the next message from the mailbox.
* @return @p true if a messages was dequeued;
* @p false if the mailbox is empty
*/
inline bool try_receive(invoke_rules& rules)
{
return self()->mailbox().try_dequeue(rules);
}
/** /**
* @brief Gets the last dequeued message from the mailbox. * @brief Gets the last dequeued message from the mailbox.
......
...@@ -32,6 +32,18 @@ class abstract_message_queue : public Super ...@@ -32,6 +32,18 @@ class abstract_message_queue : public Super
} }
} }
void dequeue(timed_invoke_rules& rules)
{
typename Super::queue_node_buffer buffer;
for (;;)
{
if (Super::dequeue_impl(rules, buffer))
{
return;
}
}
}
void dequeue(invoke_rules& rules) /*override*/ void dequeue(invoke_rules& rules) /*override*/
{ {
typename Super::queue_node_buffer buffer; typename Super::queue_node_buffer buffer;
......
...@@ -7,6 +7,8 @@ ...@@ -7,6 +7,8 @@
#include "cppa/util/single_reader_queue.hpp" #include "cppa/util/single_reader_queue.hpp"
#include "cppa/detail/abstract_message_queue.hpp" #include "cppa/detail/abstract_message_queue.hpp"
namespace cppa { class invoke_rules_base; }
namespace cppa { namespace detail { namespace cppa { namespace detail {
// blocks if single_reader_queue blocks // blocks if single_reader_queue blocks
...@@ -24,9 +26,17 @@ class blocking_message_queue_impl : public message_queue ...@@ -24,9 +26,17 @@ class blocking_message_queue_impl : public message_queue
typedef util::single_reader_queue<queue_node> queue_type; typedef util::single_reader_queue<queue_node> queue_type;
private: inline queue_type& queue()
{
return m_queue;
}
queue_type m_queue; inline const queue_type& queue() const
{
return m_queue;
}
virtual void enqueue(const message& msg) /*override*/;
protected: protected:
...@@ -42,25 +52,21 @@ class blocking_message_queue_impl : public message_queue ...@@ -42,25 +52,21 @@ class blocking_message_queue_impl : public message_queue
if (!buffer.empty()) m_queue.push_front(std::move(buffer)); if (!buffer.empty()) m_queue.push_front(std::move(buffer));
} }
// conputes the next message, returns true if the computed message // computes the next message, returns true if the computed message
// was not ignored (e.g. because it's an exit message with reason = normal) // was not ignored (e.g. because it's an exit message with reason = normal)
bool dequeue_impl(message& storage); bool dequeue_impl(message& storage);
bool dequeue_impl(invoke_rules&, queue_node_buffer&); bool dequeue_impl(invoke_rules&, queue_node_buffer&);
public: bool dequeue_impl(timed_invoke_rules&, queue_node_buffer&);
inline queue_type& queue() private:
{
return m_queue;
}
inline const queue_type& queue() const
{
return m_queue;
}
virtual void enqueue(const message& msg) /*override*/; bool dq(std::unique_ptr<queue_node>&,invoke_rules_base&,queue_node_buffer&);
queue_type m_queue;
}; };
typedef abstract_message_queue<blocking_message_queue_impl> typedef abstract_message_queue<blocking_message_queue_impl>
......
...@@ -3,9 +3,11 @@ ...@@ -3,9 +3,11 @@
#include <vector> #include <vector>
#include <cstddef> #include <cstddef>
#include <cstdint>
#include "cppa/invoke.hpp" #include "cppa/invoke.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/detail/intermediate.hpp" #include "cppa/detail/intermediate.hpp"
// forward declaration // forward declaration
...@@ -13,17 +15,66 @@ namespace cppa { class any_tuple; } ...@@ -13,17 +15,66 @@ namespace cppa { class any_tuple; }
namespace cppa { namespace detail { namespace cppa { namespace detail {
class invokable class invokable_base
{ {
invokable(const invokable&) = delete; invokable_base(const invokable_base&) = delete;
invokable& operator=(const invokable&) = delete; invokable_base& operator=(const invokable_base&) = delete;
public: public:
invokable() = default; invokable_base() = default;
virtual ~invokable(); virtual ~invokable_base();
virtual bool invoke(const any_tuple&) const = 0; virtual bool invoke(const any_tuple&) const = 0;
};
class timed_invokable : public invokable_base
{
util::duration m_timeout;
protected:
timed_invokable(const util::duration&);
public:
inline const util::duration& timeout() const
{
return m_timeout;
}
};
template<class TargetFun>
class timed_invokable_impl : public timed_invokable
{
typedef timed_invokable super;
TargetFun m_target;
public:
timed_invokable_impl(const util::duration& d, const TargetFun& tfun)
: super(d), m_target(tfun)
{
}
bool invoke(const any_tuple&) const
{
m_target();
return true;
}
};
class invokable : public invokable_base
{
public:
virtual intermediate* get_intermediate(const any_tuple&) const = 0; virtual intermediate* get_intermediate(const any_tuple&) const = 0;
}; };
...@@ -37,11 +88,6 @@ class invokable_impl : public invokable ...@@ -37,11 +88,6 @@ class invokable_impl : public invokable
public: public:
//invokable_impl(MatchFunction&& mm, TargetFun&& mt)
// : m_match(std::move(mm)), m_target(std::move(mt))
//{
//}
invokable_impl(MatchFunction&& mm, const TargetFun& mt) invokable_impl(MatchFunction&& mm, const TargetFun& mt)
: m_match(std::move(mm)), m_target(mt) : m_match(std::move(mm)), m_target(mt)
{ {
...@@ -95,11 +141,6 @@ class invokable_impl<TupleClass<>, MatchFunction, TargetFun> : public invokable ...@@ -95,11 +141,6 @@ class invokable_impl<TupleClass<>, MatchFunction, TargetFun> : public invokable
public: public:
//invokable_impl(MatchFunction&& mm, TargetFun&& mt)
// : m_match(std::move(mm)), m_target(std::move(mt))
//{
//}
invokable_impl(MatchFunction&& mm, const TargetFun& mt) invokable_impl(MatchFunction&& mm, const TargetFun& mt)
: m_match(std::move(mm)), m_target(mt) : m_match(std::move(mm)), m_target(mt)
{ {
......
...@@ -29,10 +29,11 @@ ...@@ -29,10 +29,11 @@
#ifndef RECEIVE_LOOP_HELPER_HPP #ifndef RECEIVE_LOOP_HELPER_HPP
#define RECEIVE_LOOP_HELPER_HPP #define RECEIVE_LOOP_HELPER_HPP
#include <new>
#include "cppa/receive.hpp"
#include "cppa/invoke_rules.hpp" #include "cppa/invoke_rules.hpp"
// forward declaration
namespace cppa { void receive(invoke_rules&); }
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -43,6 +44,7 @@ struct receive_while_helper ...@@ -43,6 +44,7 @@ struct receive_while_helper
receive_while_helper(Statement&& stmt) : m_stmt(std::move(stmt)) receive_while_helper(Statement&& stmt) : m_stmt(std::move(stmt))
{ {
} }
void operator()(invoke_rules& rules) void operator()(invoke_rules& rules)
{ {
while (m_stmt()) while (m_stmt())
...@@ -50,17 +52,34 @@ struct receive_while_helper ...@@ -50,17 +52,34 @@ struct receive_while_helper
receive(rules); receive(rules);
} }
} }
void operator()(invoke_rules&& rules) void operator()(invoke_rules&& rules)
{ {
invoke_rules tmp(std::move(rules)); invoke_rules tmp(std::move(rules));
(*this)(tmp); (*this)(tmp);
} }
void operator()(timed_invoke_rules& rules)
{
while (m_stmt())
{
receive(rules);
}
}
void operator()(timed_invoke_rules&& rules)
{
timed_invoke_rules tmp(std::move(rules));
(*this)(tmp);
}
template<typename Arg0, typename... Args> template<typename Arg0, typename... Args>
void operator()(invoke_rules& rules, Arg0&& arg0, Args&&... args) void operator()(invoke_rules& rules, Arg0&& arg0, Args&&... args)
{ {
(*this)(rules.splice(std::forward<Arg0>(arg0)), (*this)(rules.splice(std::forward<Arg0>(arg0)),
std::forward<Args>(args)...); std::forward<Args>(args)...);
} }
template<typename Arg0, typename... Args> template<typename Arg0, typename... Args>
void operator()(invoke_rules&& rules, Arg0&& arg0, Args&&... args) void operator()(invoke_rules&& rules, Arg0&& arg0, Args&&... args)
{ {
...@@ -73,40 +92,83 @@ struct receive_while_helper ...@@ -73,40 +92,83 @@ struct receive_while_helper
class do_receive_helper class do_receive_helper
{ {
invoke_rules m_rules; bool m_timed;
union
{
invoke_rules m_rules;
timed_invoke_rules m_trules;
};
inline void init(timed_invoke_rules&& ti_rules)
{
m_timed = true;
m_rules.~invoke_rules();
new (&m_trules) timed_invoke_rules(std::move(ti_rules));
}
inline void init(invoke_rules&) { } inline void init(invoke_rules&) { }
template<typename Arg0, typename... Args> template<typename Arg0, typename... Args>
inline void init(invoke_rules& rules, Arg0&& arg0, Args&&... args) inline void init(invoke_rules& rules, Arg0&& arg0, Args&&... args)
{ {
init(rules.splice(arg0), std::forward<Args>(args)...); init(rules.splice(std::forward<Arg0>(arg0)),
std::forward<Args>(args)...);
} }
public: public:
//template<typename... Args>
//do_receive_helper(invoke_rules& rules, Args&&... args) : m_rules(std::move(rules))
//{
// init(m_rules, std::forward<Args>(args)...);
//}
template<typename... Args> template<typename... Args>
do_receive_helper(invoke_rules&& rules, Args&&... args) : m_rules(std::move(rules)) do_receive_helper(invoke_rules&& rules, Args&&... args) : m_timed(false)
{ {
new (&m_rules) invoke_rules(std::move(rules));
init(m_rules, std::forward<Args>(args)...); init(m_rules, std::forward<Args>(args)...);
} }
do_receive_helper(timed_invoke_rules&& rules) : m_timed(true)
{
new (&m_trules) timed_invoke_rules(std::move(rules));
}
do_receive_helper(do_receive_helper&& other) : m_timed(other.m_timed)
{
if (other.m_timed)
{
new (&m_trules) timed_invoke_rules(std::move(other.m_trules));
}
else
{
new (&m_rules) invoke_rules(std::move(other.m_rules));
}
}
~do_receive_helper()
{
if (m_timed) m_trules.~timed_invoke_rules();
else m_rules.~invoke_rules();
}
template<typename Statement> template<typename Statement>
void until(Statement&& stmt) void until(Statement&& stmt)
{ {
static_assert(std::is_same<bool, decltype(stmt())>::value, static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean"); "functor or function does not return a boolean");
do if (m_timed)
{
do
{
receive(m_trules);
}
while (stmt() == false);
}
else
{ {
receive(m_rules); do
{
receive(m_rules);
}
while (stmt() == false);
} }
while (stmt() == false);
} }
}; };
......
#ifndef YIELDING_MESSAGE_QUEUE_HPP #ifndef YIELDING_MESSAGE_QUEUE_HPP
#define YIELDING_MESSAGE_QUEUE_HPP #define YIELDING_MESSAGE_QUEUE_HPP
#include <cstdint>
#include "cppa/message.hpp" #include "cppa/message.hpp"
#include "cppa/message_queue.hpp" #include "cppa/message_queue.hpp"
#include "cppa/util/shared_spinlock.hpp" #include "cppa/util/shared_spinlock.hpp"
...@@ -9,6 +11,8 @@ ...@@ -9,6 +11,8 @@
#include "cppa/util/single_reader_queue.hpp" #include "cppa/util/single_reader_queue.hpp"
#include "cppa/detail/abstract_message_queue.hpp" #include "cppa/detail/abstract_message_queue.hpp"
namespace cppa { class invoke_rules_base; }
namespace cppa { namespace detail { namespace cppa { namespace detail {
class scheduled_actor; class scheduled_actor;
...@@ -34,12 +38,31 @@ class yielding_message_queue_impl : public message_queue ...@@ -34,12 +38,31 @@ class yielding_message_queue_impl : public message_queue
queue_node(const message& from); queue_node(const message& from);
}; };
bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id;
scheduled_actor* m_parent; scheduled_actor* m_parent;
std::atomic<int> m_state; std::atomic<int> m_state;
util::single_reader_queue<queue_node> m_queue; util::single_reader_queue<queue_node> m_queue;
void yield_until_not_empty(); void yield_until_not_empty();
enum dq_result
{
done,
indeterminate,
timeout_occured
};
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
filter_result filter_msg(const message& msg);
protected: protected:
typedef util::singly_linked_list<queue_node> queue_node_buffer; typedef util::singly_linked_list<queue_node> queue_node_buffer;
...@@ -54,12 +77,18 @@ class yielding_message_queue_impl : public message_queue ...@@ -54,12 +77,18 @@ class yielding_message_queue_impl : public message_queue
if (!buffer.empty()) m_queue.push_front(std::move(buffer)); if (!buffer.empty()) m_queue.push_front(std::move(buffer));
} }
auto dq(std::unique_ptr<queue_node>&,
invoke_rules_base&,
queue_node_buffer&) -> dq_result;
// conputes the next message, returns true if the computed message // conputes the next message, returns true if the computed message
// was not ignored (e.g. because it's an exit message with reason = normal) // was not ignored (e.g. because it's an exit message with reason = normal)
bool dequeue_impl(message& storage); bool dequeue_impl(message& storage);
bool dequeue_impl(invoke_rules&, queue_node_buffer&); bool dequeue_impl(invoke_rules&, queue_node_buffer&);
bool dequeue_impl(timed_invoke_rules&, queue_node_buffer&);
public: public:
yielding_message_queue_impl(scheduled_actor* parent); yielding_message_queue_impl(scheduled_actor* parent);
......
...@@ -32,42 +32,110 @@ ...@@ -32,42 +32,110 @@
#include <list> #include <list>
#include <memory> #include <memory>
namespace cppa { namespace detail { class invokable; class intermediate; } } namespace cppa { namespace detail {
class invokable;
class intermediate;
class timed_invokable;
typedef std::unique_ptr<detail::invokable> invokable_ptr;
typedef std::unique_ptr<detail::timed_invokable> timed_invokable_ptr;
} } // namespace cppa::detail
namespace cppa { namespace util { class duration; } }
namespace cppa { namespace cppa {
class any_tuple; class any_tuple;
class invoke_rules;
typedef std::list<detail::invokable_ptr> invokable_list;
class invoke_rules_base
{
protected:
invokable_list m_list;
invoke_rules_base() = default;
invoke_rules_base(invokable_list&& ilist);
invoke_rules_base(invoke_rules_base&& other);
public:
virtual ~invoke_rules_base();
bool operator()(const any_tuple& t) const;
detail::intermediate* get_intermediate(const any_tuple& t) const;
};
class timed_invoke_rules : public invoke_rules_base
{
struct invoke_rules typedef invoke_rules_base super;
friend class invoke_rules;
timed_invoke_rules() = delete;
timed_invoke_rules(const timed_invoke_rules&) = delete;
timed_invoke_rules& operator=(const timed_invoke_rules&) = delete;
timed_invoke_rules(invokable_list&& prepended_list,
timed_invoke_rules&& other);
public:
timed_invoke_rules(timed_invoke_rules&& arg);
timed_invoke_rules(detail::timed_invokable_ptr&& arg);
const util::duration& timeout() const;
void handle_timeout() const;
private:
detail::timed_invokable_ptr m_ti;
};
class invoke_rules : public invoke_rules_base
{ {
typedef std::list< std::unique_ptr<detail::invokable> > list_type; typedef invoke_rules_base super;
list_type m_list; friend class timed_invoke_rules;
invoke_rules(const invoke_rules&) = delete; invoke_rules(const invoke_rules&) = delete;
invoke_rules& operator=(const invoke_rules&) = delete; invoke_rules& operator=(const invoke_rules&) = delete;
invoke_rules(list_type&& ll);
public: public:
invoke_rules() = default; invoke_rules() = default;
invoke_rules(invoke_rules&& other); invoke_rules(invoke_rules&& other);
invoke_rules(detail::invokable* arg);
invoke_rules(std::unique_ptr<detail::invokable>&& arg); invoke_rules(std::unique_ptr<detail::invokable>&& arg);
bool operator()(const any_tuple& t) const;
detail::intermediate* get_intermediate(const any_tuple& t) const;
invoke_rules& splice(invoke_rules&& other); invoke_rules& splice(invoke_rules&& other);
timed_invoke_rules splice(timed_invoke_rules&& other);
invoke_rules operator,(invoke_rules&& other); invoke_rules operator,(invoke_rules&& other);
timed_invoke_rules operator,(timed_invoke_rules&& other);
private:
invoke_rules(invokable_list&& ll);
invoke_rules& splice(invokable_list&& ilist);
}; };
} // namespace cppa } // namespace cppa
......
...@@ -37,6 +37,8 @@ class message ...@@ -37,6 +37,8 @@ class message
intrusive_ptr<msg_content> m_content; intrusive_ptr<msg_content> m_content;
static msg_content* s_dummy;
public: public:
template<typename... Args> template<typename... Args>
...@@ -75,6 +77,10 @@ class message ...@@ -75,6 +77,10 @@ class message
return m_content->data; return m_content->data;
} }
bool empty() const;
static msg_content* create_dummy();
}; };
template<typename... Args> template<typename... Args>
......
...@@ -8,6 +8,7 @@ namespace cppa { ...@@ -8,6 +8,7 @@ namespace cppa {
// forward declaration // forward declaration
class invoke_rules; class invoke_rules;
class timed_invoke_rules;
class message_queue : public ref_counted class message_queue : public ref_counted
{ {
...@@ -24,6 +25,8 @@ class message_queue : public ref_counted ...@@ -24,6 +25,8 @@ class message_queue : public ref_counted
virtual void enqueue(const message&) = 0; virtual void enqueue(const message&) = 0;
virtual const message& dequeue() = 0; virtual const message& dequeue() = 0;
virtual void dequeue(invoke_rules&) = 0; virtual void dequeue(invoke_rules&) = 0;
virtual void dequeue(timed_invoke_rules&) = 0;
virtual bool try_dequeue(message&) = 0; virtual bool try_dequeue(message&) = 0;
virtual bool try_dequeue(invoke_rules&) = 0; virtual bool try_dequeue(invoke_rules&) = 0;
......
#ifndef ON_HPP #ifndef ON_HPP
#define ON_HPP #define ON_HPP
#include <chrono>
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
#include "cppa/match.hpp" #include "cppa/match.hpp"
#include "cppa/invoke.hpp" #include "cppa/invoke.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/invoke_rules.hpp" #include "cppa/invoke_rules.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/util/eval_first_n.hpp" #include "cppa/util/eval_first_n.hpp"
#include "cppa/util/filter_type_list.hpp" #include "cppa/util/filter_type_list.hpp"
...@@ -17,6 +20,26 @@ ...@@ -17,6 +20,26 @@
namespace cppa { namespace detail { namespace cppa { namespace detail {
class timed_invoke_rule_builder
{
util::duration m_timeout;
public:
constexpr timed_invoke_rule_builder(const util::duration& d) : m_timeout(d)
{
}
template<typename F>
timed_invoke_rules operator>>(F&& f)
{
typedef timed_invokable_impl<F> impl;
return timed_invokable_ptr(new impl(m_timeout, std::forward<F>(f)));
}
};
template<typename... TypeList> template<typename... TypeList>
class invoke_rule_builder class invoke_rule_builder
{ {
...@@ -39,10 +62,10 @@ class invoke_rule_builder ...@@ -39,10 +62,10 @@ class invoke_rule_builder
invoke_rule_builder(match_function&& fun) : m_fun(std::move(fun)) { } invoke_rule_builder(match_function&& fun) : m_fun(std::move(fun)) { }
template<typename F> template<typename F>
cppa::invoke_rules operator>>(F&& f) invoke_rules operator>>(F&& f)
{ {
typedef invokable_impl<tuple_view_type, match_function, F> impl; typedef invokable_impl<tuple_view_type, match_function, F> impl;
return { new impl(std::move(m_fun), std::forward<F>(f)) }; return invokable_ptr(new impl(std::move(m_fun), std::forward<F>(f)));
} }
}; };
...@@ -170,6 +193,13 @@ detail::invoke_rule_builder<atom_value, atom_value, atom_value, ...@@ -170,6 +193,13 @@ detail::invoke_rule_builder<atom_value, atom_value, atom_value,
}; };
} }
template<class Rep, class Period>
inline constexpr detail::timed_invoke_rule_builder
after(const std::chrono::duration<Rep, Period>& d)
{
return { util::duration(d) };
}
inline detail::invoke_rule_builder<any_type*> others() inline detail::invoke_rule_builder<any_type*> others()
{ {
return return
......
#ifndef RECEIVE_HPP
#define RECEIVE_HPP
#include "cppa/context.hpp"
namespace cppa {
/**
* @brief Dequeues the next message from the mailbox.
*/
inline const message& receive()
{
return self()->mailbox().dequeue();
}
/**
* @brief Dequeues the next message from the mailbox that's matched
* by @p rules and executes the corresponding callback.
*/
inline void receive(invoke_rules& rules)
{
self()->mailbox().dequeue(rules);
}
inline void receive(timed_invoke_rules& rules)
{
self()->mailbox().dequeue(rules);
}
inline void receive(timed_invoke_rules&& rules)
{
timed_invoke_rules tmp(std::move(rules));
self()->mailbox().dequeue(tmp);
}
inline void receive(invoke_rules&& rules)
{
invoke_rules tmp(std::move(rules));
self()->mailbox().dequeue(tmp);
}
template<typename Head, typename... Tail>
void receive(invoke_rules&& rules, Head&& head, Tail&&... tail)
{
invoke_rules tmp(std::move(rules));
receive(tmp.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
template<typename Head, typename... Tail>
void receive(invoke_rules& rules, Head&& head, Tail&&... tail)
{
receive(rules.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
/**
* @brief Tries to dequeue the next message from the mailbox.
* @return @p true if a messages was dequeued;
* @p false if the mailbox is empty
*/
inline bool try_receive(message& msg)
{
return self()->mailbox().try_dequeue(msg);
}
/**
* @brief Tries to dequeue the next message from the mailbox.
* @return @p true if a messages was dequeued;
* @p false if the mailbox is empty
*/
inline bool try_receive(invoke_rules& rules)
{
return self()->mailbox().try_dequeue(rules);
}
} // namespace cppa
#endif // RECEIVE_HPP
...@@ -4,6 +4,8 @@ ...@@ -4,6 +4,8 @@
#include <chrono> #include <chrono>
#include <cstdint> #include <cstdint>
#include <boost/thread/thread_time.hpp>
namespace cppa { namespace util { namespace cppa { namespace util {
enum class time_unit : std::uint32_t enum class time_unit : std::uint32_t
...@@ -64,4 +66,10 @@ inline bool operator!=(const duration& lhs, const duration& rhs) ...@@ -64,4 +66,10 @@ inline bool operator!=(const duration& lhs, const duration& rhs)
} } // namespace cppa::util } } // namespace cppa::util
namespace boost {
system_time& operator+=(system_time& lhs, const cppa::util::duration& d);
} // namespace boost
#endif // DURATION_HPP #endif // DURATION_HPP
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <boost/thread.hpp>
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
#include "cppa/match.hpp" #include "cppa/match.hpp"
#include "cppa/context.hpp" #include "cppa/context.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/invoke_rules.hpp" #include "cppa/invoke_rules.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/util/singly_linked_list.hpp" #include "cppa/util/singly_linked_list.hpp"
#include "cppa/detail/intermediate.hpp" #include "cppa/detail/intermediate.hpp"
...@@ -68,16 +71,13 @@ bool blocking_message_queue_impl::dequeue_impl(message& storage) ...@@ -68,16 +71,13 @@ bool blocking_message_queue_impl::dequeue_impl(message& storage)
return true; return true;
} }
bool blocking_message_queue_impl::dequeue_impl(invoke_rules& rules, bool blocking_message_queue_impl::dq(std::unique_ptr<queue_node>& node,
queue_node_buffer& buffer) invoke_rules_base& rules,
queue_node_buffer& buffer)
{ {
std::unique_ptr<queue_node> node(m_queue.pop()); if (!m_trap_exit && throw_on_exit(node->msg) == normal_exit_signal)
if (!m_trap_exit)
{ {
if (throw_on_exit(node->msg) == normal_exit_signal) return false;
{
return false;
}
} }
std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg.content())); std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg.content()));
if (imd) if (imd)
...@@ -93,6 +93,33 @@ bool blocking_message_queue_impl::dequeue_impl(invoke_rules& rules, ...@@ -93,6 +93,33 @@ bool blocking_message_queue_impl::dequeue_impl(invoke_rules& rules,
buffer.push_back(node.release()); buffer.push_back(node.release());
return false; return false;
} }
}
bool blocking_message_queue_impl::dequeue_impl(timed_invoke_rules& rules,
queue_node_buffer& buffer)
{
std::unique_ptr<queue_node> node(m_queue.try_pop());
if (!node)
{
boost::system_time timeout = boost::get_system_time();
timeout += rules.timeout();
node.reset(m_queue.try_pop(timeout));
if (!node)
{
if (!buffer.empty()) m_queue.push_front(std::move(buffer));
rules.handle_timeout();
return true;
}
}
return dq(node, rules, buffer);
}
bool blocking_message_queue_impl::dequeue_impl(invoke_rules& rules,
queue_node_buffer& buffer)
{
std::unique_ptr<queue_node> node(m_queue.pop());
return dq(node, rules, buffer);
} }
} } // namespace hamcast::detail } } // namespace hamcast::detail
...@@ -111,4 +111,13 @@ void receive_loop(invoke_rules& rules) ...@@ -111,4 +111,13 @@ void receive_loop(invoke_rules& rules)
} }
} }
void receive_loop(timed_invoke_rules& rules)
{
auto& mbox = self()->mailbox();
for (;;)
{
mbox.dequeue(rules);
}
}
} // namespace cppa } // namespace cppa
...@@ -18,3 +18,28 @@ bool operator==(const duration& lhs, const duration& rhs) ...@@ -18,3 +18,28 @@ bool operator==(const duration& lhs, const duration& rhs)
} }
} } // namespace cppa::util } } // namespace cppa::util
namespace boost {
system_time& operator+=(system_time& lhs, const cppa::util::duration& d)
{
switch (d.unit)
{
case cppa::util::time_unit::seconds:
lhs += posix_time::seconds(d.count);
break;
case cppa::util::time_unit::milliseconds:
lhs += posix_time::milliseconds(d.count);
break;
case cppa::util::time_unit::microseconds:
lhs += posix_time::microseconds(d.count);
break;
default: break;
}
return lhs;
}
} // namespace boost
...@@ -2,8 +2,12 @@ ...@@ -2,8 +2,12 @@
namespace cppa { namespace detail { namespace cppa { namespace detail {
invokable::~invokable() invokable_base::~invokable_base()
{ {
} }
} } timed_invokable::timed_invokable(const util::duration& d) : m_timeout(d)
{
}
} } // namespace cppa::detail
...@@ -28,30 +28,28 @@ ...@@ -28,30 +28,28 @@
#include "cppa/invoke.hpp" #include "cppa/invoke.hpp"
#include "cppa/invoke_rules.hpp" #include "cppa/invoke_rules.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/detail/invokable.hpp" #include "cppa/detail/invokable.hpp"
namespace cppa { namespace cppa {
invoke_rules::invoke_rules(list_type&& ll) : m_list(std::move(ll)) // invoke_rules_base
{
}
invoke_rules::invoke_rules(invoke_rules&& other) invoke_rules_base::invoke_rules_base(invoke_rules_base&& other)
: m_list(std::move(other.m_list)) : m_list(std::move(other.m_list))
{ {
} }
invoke_rules::invoke_rules(detail::invokable* arg) invoke_rules_base::invoke_rules_base(invokable_list&& ilist)
: m_list(std::move(ilist))
{ {
if (arg) m_list.push_back(std::unique_ptr<detail::invokable>(arg));
} }
invoke_rules::invoke_rules(std::unique_ptr<detail::invokable>&& arg) invoke_rules_base::~invoke_rules_base()
{ {
if (arg) m_list.push_back(std::move(arg));
} }
bool invoke_rules::operator()(const any_tuple& t) const bool invoke_rules_base::operator()(const any_tuple& t) const
{ {
for (auto i = m_list.begin(); i != m_list.end(); ++i) for (auto i = m_list.begin(); i != m_list.end(); ++i)
{ {
...@@ -60,7 +58,8 @@ bool invoke_rules::operator()(const any_tuple& t) const ...@@ -60,7 +58,8 @@ bool invoke_rules::operator()(const any_tuple& t) const
return false; return false;
} }
detail::intermediate* invoke_rules::get_intermediate(const any_tuple& t) const detail::intermediate*
invoke_rules_base::get_intermediate(const any_tuple& t) const
{ {
detail::intermediate* result; detail::intermediate* result;
for (auto i = m_list.begin(); i != m_list.end(); ++i) for (auto i = m_list.begin(); i != m_list.end(); ++i)
...@@ -71,17 +70,71 @@ detail::intermediate* invoke_rules::get_intermediate(const any_tuple& t) const ...@@ -71,17 +70,71 @@ detail::intermediate* invoke_rules::get_intermediate(const any_tuple& t) const
return nullptr; return nullptr;
} }
invoke_rules& invoke_rules::splice(invoke_rules&& other) // timed_invoke_rules
timed_invoke_rules::timed_invoke_rules(timed_invoke_rules&& other)
: super(std::move(other)), m_ti(std::move(other.m_ti))
{ {
m_list.splice(m_list.end(), other.m_list); }
timed_invoke_rules::timed_invoke_rules(detail::timed_invokable_ptr&& arg)
: super(), m_ti(std::move(arg))
{
}
timed_invoke_rules::timed_invoke_rules(invokable_list&& lhs,
timed_invoke_rules&& rhs)
: super(std::move(lhs)), m_ti(std::move(rhs.m_ti))
{
m_list.splice(m_list.begin(), rhs.m_list);
}
const util::duration& timed_invoke_rules::timeout() const
{
return m_ti->timeout();
}
void timed_invoke_rules::handle_timeout() const
{
// safe, because timed_invokable ignores the given argument
m_ti->invoke(*static_cast<any_tuple*>(nullptr));
}
// invoke_rules
invoke_rules::invoke_rules(invokable_list&& ll) : super(std::move(ll))
{
}
invoke_rules::invoke_rules(invoke_rules&& other) : super(std::move(other))
{
}
invoke_rules::invoke_rules(std::unique_ptr<detail::invokable>&& arg)
{
if (arg) m_list.push_back(std::move(arg));
}
invoke_rules& invoke_rules::splice(invokable_list&& ilist)
{
m_list.splice(m_list.end(), ilist);
return *this; return *this;
} }
invoke_rules& invoke_rules::splice(invoke_rules&& other)
{
return splice(std::move(other.m_list));
}
timed_invoke_rules invoke_rules::splice(timed_invoke_rules&& other)
{
return timed_invoke_rules(std::move(m_list), std::move(other));
}
invoke_rules invoke_rules::operator,(invoke_rules&& other) invoke_rules invoke_rules::operator,(invoke_rules&& other)
{ {
m_list.splice(m_list.end(), other.m_list); m_list.splice(m_list.end(), other.m_list);
return std::move(m_list); return std::move(m_list);
} }
} // namespace cppa } // namespace cppa
...@@ -2,6 +2,15 @@ ...@@ -2,6 +2,15 @@
namespace cppa { namespace cppa {
message::msg_content* message::create_dummy()
{
msg_content* result = new msg_content(0, 0, tuple<int>(0));
result->ref();
return result;
}
message::msg_content* message::s_dummy = message::create_dummy();
message::message(const actor_ptr& from, message::message(const actor_ptr& from,
const channel_ptr& to, const channel_ptr& to,
const any_tuple& ut) const any_tuple& ut)
...@@ -16,8 +25,13 @@ message::message(const actor_ptr& from, ...@@ -16,8 +25,13 @@ message::message(const actor_ptr& from,
{ {
} }
message::message() : m_content(new msg_content(0, 0, tuple<int>(0))) message::message() : m_content(s_dummy)
{
}
bool message::empty() const
{ {
return m_content.get() == s_dummy;
} }
bool operator==(const message& lhs, const message& rhs) bool operator==(const message& lhs, const message& rhs)
......
...@@ -6,8 +6,10 @@ ...@@ -6,8 +6,10 @@
#include "cppa/match.hpp" #include "cppa/match.hpp"
#include "cppa/context.hpp" #include "cppa/context.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/invoke_rules.hpp" #include "cppa/invoke_rules.hpp"
#include "cppa/detail/intermediate.hpp" #include "cppa/detail/intermediate.hpp"
#include "cppa/detail/scheduled_actor.hpp" #include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/yield_interface.hpp" #include "cppa/detail/yield_interface.hpp"
...@@ -22,31 +24,6 @@ namespace { ...@@ -22,31 +24,6 @@ namespace {
typedef std::lock_guard<cppa::util::shared_spinlock> exclusive_guard; typedef std::lock_guard<cppa::util::shared_spinlock> exclusive_guard;
typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard; typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard;
enum yield_on_exit_result
{
normal_exit_signal,
not_an_exit_signal
};
yield_on_exit_result yield_on_exit_msg(const cppa::message& msg)
{
if (cppa::match<cppa::atom_value, std::uint32_t>(msg.content(), nullptr,
cppa::atom(":Exit")))
{
auto reason = *reinterpret_cast<const std::uint32_t*>(msg.content().at(1));
if (reason != cppa::exit_reason::normal)
{
// yields
cppa::self()->quit(reason);
}
else
{
return normal_exit_signal;
}
}
return not_an_exit_signal;
}
} // namespace <anonmyous> } // namespace <anonmyous>
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -57,8 +34,8 @@ yielding_message_queue_impl::queue_node::queue_node(const message& from) ...@@ -57,8 +34,8 @@ yielding_message_queue_impl::queue_node::queue_node(const message& from)
} }
yielding_message_queue_impl::yielding_message_queue_impl(scheduled_actor* p) yielding_message_queue_impl::yielding_message_queue_impl(scheduled_actor* p)
: m_parent(p) : m_has_pending_timeout_request(false), m_active_timeout_id(0)
, m_state(ready) , m_parent(p), m_state(ready)
{ {
} }
...@@ -66,6 +43,41 @@ yielding_message_queue_impl::~yielding_message_queue_impl() ...@@ -66,6 +43,41 @@ yielding_message_queue_impl::~yielding_message_queue_impl()
{ {
} }
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
yielding_message_queue_impl::filter_result
yielding_message_queue_impl::filter_msg(const message& msg)
{
if ( m_trap_exit == false
&& match<atom_value, std::uint32_t>(msg.content(),
nullptr,
atom(":Exit")))
{
auto why = *reinterpret_cast<const std::uint32_t*>(msg.content().at(1));
if (why != cppa::exit_reason::normal)
{
// yields
cppa::self()->quit(why);
}
return normal_exit_signal;
}
if (match<atom_value, std::uint32_t>(msg.content(),
nullptr,
atom(":Timeout")))
{
auto id = *reinterpret_cast<const std::uint32_t*>(msg.content().at(1));
return (id == m_active_timeout_id) ? timeout_message
: expired_timeout_message;
}
return ordinary_message;
}
void yielding_message_queue_impl::enqueue(const message& msg) void yielding_message_queue_impl::enqueue(const message& msg)
{ {
if (m_queue._push_back(new queue_node(msg))) if (m_queue._push_back(new queue_node(msg)))
...@@ -122,30 +134,31 @@ bool yielding_message_queue_impl::dequeue_impl(message& storage) ...@@ -122,30 +134,31 @@ bool yielding_message_queue_impl::dequeue_impl(message& storage)
{ {
yield_until_not_empty(); yield_until_not_empty();
std::unique_ptr<queue_node> node(m_queue.pop()); std::unique_ptr<queue_node> node(m_queue.pop());
if (!m_trap_exit) if (filter_msg(node->msg) != ordinary_message)
{ {
if (yield_on_exit_msg(node->msg) == normal_exit_signal) // dequeue next message
{ return false;
// exit_reason::normal is ignored by default,
// dequeue next message
return false;
}
} }
storage = std::move(node->msg); storage = std::move(node->msg);
return true; return true;
} }
bool yielding_message_queue_impl::dequeue_impl(invoke_rules& rules, yielding_message_queue_impl::dq_result
queue_node_buffer& buffer) yielding_message_queue_impl::dq(std::unique_ptr<queue_node>& node,
invoke_rules_base& rules,
queue_node_buffer& buffer)
{ {
yield_until_not_empty(); switch (filter_msg(node->msg))
std::unique_ptr<queue_node> node(m_queue.pop());
if (!m_trap_exit)
{ {
if (yield_on_exit_msg(node->msg) == normal_exit_signal) case normal_exit_signal:
{ case expired_timeout_message:
return false; // discard message
} return indeterminate;
case timeout_message:
return timeout_occured;
default: break;
} }
std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg.content())); std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg.content()));
if (imd) if (imd)
...@@ -154,13 +167,53 @@ bool yielding_message_queue_impl::dequeue_impl(invoke_rules& rules, ...@@ -154,13 +167,53 @@ bool yielding_message_queue_impl::dequeue_impl(invoke_rules& rules,
// restore mailbox before invoking imd // restore mailbox before invoking imd
if (!buffer.empty()) m_queue.push_front(std::move(buffer)); if (!buffer.empty()) m_queue.push_front(std::move(buffer));
imd->invoke(); imd->invoke();
return true; return done;
} }
else else
{ {
buffer.push_back(node.release()); buffer.push_back(node.release());
return false; return indeterminate;
}
}
bool yielding_message_queue_impl::dequeue_impl(timed_invoke_rules& rules, queue_node_buffer& buffer)
{
if (m_queue.empty() && !m_has_pending_timeout_request)
{
get_scheduler()->future_send(self(), rules.timeout(),
atom(":Timeout"), ++m_active_timeout_id);
m_has_pending_timeout_request = true;
} }
yield_until_not_empty();
std::unique_ptr<queue_node> node(m_queue.pop());
switch (dq(node, rules, buffer))
{
case done:
if (m_has_pending_timeout_request)
{
// expire pending request
++m_active_timeout_id;
m_has_pending_timeout_request = false;
}
return true;
case timeout_occured:
rules.handle_timeout();
m_has_pending_timeout_request = false;
return true;
case indeterminate:
return false;
}
return false;
}
bool yielding_message_queue_impl::dequeue_impl(invoke_rules& rules,
queue_node_buffer& buffer)
{
yield_until_not_empty();
std::unique_ptr<queue_node> node(m_queue.pop());
return dq(node, rules, buffer) != indeterminate;
} }
} } // namespace cppa::detail } } // namespace cppa::detail
...@@ -36,7 +36,15 @@ size_t test__local_group() ...@@ -36,7 +36,15 @@ size_t test__local_group()
//receive_until([&result]() { return result == 10; }) //receive_until([&result]() { return result == 10; })
do_receive do_receive
( (
on<int>() >> [&result](int value) { result += value; } on<int>() >> [&result](int value)
{
result += value;
},
after(std::chrono::seconds(2)) >> [&]()
{
CPPA_CHECK(false);
}
) )
.until([&result]() { return result == 10; }); .until([&result]() { return result == 10; });
await_all_others_done(); await_all_others_done();
......
...@@ -18,6 +18,38 @@ using std::endl; ...@@ -18,6 +18,38 @@ using std::endl;
using namespace cppa; using namespace cppa;
void testee1()
{
receive_loop
(
others() >> []()
{
message msg = last_received();
actor_ptr sender = msg.sender();
sender->enqueue(message(self(), sender, msg.content()));
},
after(std::chrono::milliseconds(10)) >> []()
{
quit(exit_reason::user_defined);
}
);
}
void testee2(actor_ptr other)
{
link(other);
send(other, std::uint32_t(1));
receive_loop
(
on<std::uint32_t>() >> [](std::uint32_t sleep_time)
{
// "sleep" for sleep_time milliseconds
receive(after(std::chrono::milliseconds(sleep_time)) >> []() {});
reply(sleep_time * 2);
}
);
}
size_t test__spawn() size_t test__spawn()
{ {
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
...@@ -30,12 +62,13 @@ size_t test__spawn() ...@@ -30,12 +62,13 @@ size_t test__spawn()
auto pong_actor = spawn(pong, spawn(ping)); auto pong_actor = spawn(pong, spawn(ping));
monitor(pong_actor); monitor(pong_actor);
link(pong_actor); link(pong_actor);
monitor(spawn(testee2, spawn(testee1)));
int i = 0; int i = 0;
int flags = 0; int flags = 0;
get_scheduler()->future_send(self(), std::chrono::seconds(1), get_scheduler()->future_send(self(), std::chrono::seconds(1),
atom("FooBar")); atom("FooBar"));
// wait for :Down and :Exit messages of pong // wait for :Down and :Exit messages of pong
receive_while([&i]() { return ++i <= 3; }) receive_while([&i]() { return ++i <= 4; })
( (
on<atom(":Exit"), std::uint32_t>() >> [&](std::uint32_t reason) on<atom(":Exit"), std::uint32_t>() >> [&](std::uint32_t reason)
{ {
...@@ -46,21 +79,34 @@ size_t test__spawn() ...@@ -46,21 +79,34 @@ size_t test__spawn()
on<atom(":Down"), std::uint32_t>() >> [&](std::uint32_t reason) on<atom(":Down"), std::uint32_t>() >> [&](std::uint32_t reason)
{ {
CPPA_CHECK_EQUAL(reason, exit_reason::user_defined); CPPA_CHECK_EQUAL(reason, exit_reason::user_defined);
CPPA_CHECK_EQUAL(last_received().sender(), pong_actor); if (last_received().sender() == pong_actor)
flags |= 0x02; {
flags |= 0x02;
}
else
{
flags |= 0x04;
}
}, },
on<atom("FooBar")>() >> [&]() on<atom("FooBar")>() >> [&]()
{ {
flags |= 0x04; flags |= 0x08;
}, },
others() >> [&]() others() >> [&]()
{ {
report_unexpected(); report_unexpected();
CPPA_CHECK(false);
},
after(std::chrono::seconds(5)) >> [&]()
{
cout << "!!! TIMEOUT !!!" << endl;
CPPA_CHECK(false);
} }
); );
// wait for termination of all spawned actors // wait for termination of all spawned actors
await_all_others_done(); await_all_others_done();
CPPA_CHECK_EQUAL(flags, 0x07); CPPA_CHECK_EQUAL(flags, 0x0F);
// mailbox has to be empty // mailbox has to be empty
message msg; message msg;
while (try_receive(msg)) while (try_receive(msg))
......
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