Commit f5fc8ca8 authored by neverlord's avatar neverlord

moving enqueue

parent be73e593
......@@ -21,6 +21,8 @@ class actor_proxy : public detail::abstract_actor<actor>
actor_proxy(std::uint32_t mid, const process_information_ptr& parent);
void enqueue(any_tuple&& msg);
void enqueue(const any_tuple& msg);
void link_to(intrusive_ptr<actor>& other);
......
......@@ -9,12 +9,18 @@
namespace cppa {
template<typename... Types>
tuple_view<Types...> tuple_cast(any_tuple&& tup);
/**
* @brief Describes a fixed-length tuple with elements of any type.
*/
class any_tuple
{
template<typename... Types>
friend tuple_view<Types...> tuple_cast(any_tuple&& tup);
cow_ptr<detail::abstract_tuple> m_vals;
explicit any_tuple(const cow_ptr<detail::abstract_tuple>& vals);
......@@ -92,6 +98,17 @@ inline T& any_tuple::get_mutable_as(size_t p)
return *reinterpret_cast<T*>(mutable_at(p));
}
template<typename... Types>
tuple_view<Types...> tuple_cast(const any_tuple& tup)
{
return tuple_view<Types...>::from(tup.vals());
}
template<typename... Types>
tuple_view<Types...> tuple_cast(any_tuple&& tup)
{
return tuple_view<Types...>::from(std::move(tup.m_vals));
}
} // namespace cppa
......
......@@ -36,6 +36,8 @@ class channel : public ref_counted
*/
virtual void enqueue(const any_tuple& msg) = 0;
virtual void enqueue(any_tuple&& msg) = 0;
};
/**
......
......@@ -346,7 +346,7 @@ detail::do_receive_helper do_receive(Args&&... args)
* @brief Gets the last dequeued message from the mailbox.
* @returns The last dequeued message from the mailbox.
*/
inline const any_tuple& last_received()
inline any_tuple& last_received()
{
return self()->mailbox().last_dequeued();
}
......
......@@ -21,7 +21,8 @@ class blocking_message_queue_impl : public message_queue
{
queue_node* next;
any_tuple msg;
queue_node(const any_tuple& from);
queue_node(any_tuple&& content);
queue_node(const any_tuple& content);
};
typedef util::single_reader_queue<queue_node> queue_type;
......@@ -36,6 +37,8 @@ class blocking_message_queue_impl : public message_queue
return m_queue;
}
virtual void enqueue(any_tuple&& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
protected:
......
......@@ -35,7 +35,8 @@ class yielding_message_queue_impl : public message_queue
{
queue_node* next;
any_tuple msg;
queue_node(const any_tuple& from);
queue_node(any_tuple&& content);
queue_node(const any_tuple& content);
};
bool m_has_pending_timeout_request;
......@@ -46,6 +47,8 @@ class yielding_message_queue_impl : public message_queue
void yield_until_not_empty();
void enqueue_node(queue_node*);
enum dq_result
{
done,
......@@ -95,6 +98,8 @@ class yielding_message_queue_impl : public message_queue
~yielding_message_queue_impl();
virtual void enqueue(any_tuple&& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
};
......
......@@ -43,6 +43,8 @@ class local_actor : public actor
*/
virtual void enqueue(const any_tuple& msg) /*override*/;
virtual void enqueue(any_tuple&& msg);
inline bool trap_exit() const;
inline void trap_exit(bool new_value);
......
......@@ -28,6 +28,8 @@ class message_queue : public ref_counted
*/
message_queue();
virtual void enqueue(any_tuple&& msg) = 0;
/**
* @brief Enqueues a new element to the message queue.
* @param msg The new message.
......@@ -86,7 +88,7 @@ class message_queue : public ref_counted
* by a dequeue() or try_dequeue() member function call.
* @warning Call only from the owner of the queue.
*/
inline const any_tuple& last_dequeued() const;
inline any_tuple& last_dequeued();
};
......@@ -104,7 +106,7 @@ inline void message_queue::trap_exit(bool value)
m_trap_exit = value;
}
inline const any_tuple& message_queue::last_dequeued() const
inline any_tuple& message_queue::last_dequeued()
{
return m_last_dequeued;
}
......
......@@ -41,6 +41,16 @@ class tuple_view
typedef cow_ptr<detail::abstract_tuple> vals_t;
static tuple_view from(const vals_t& vals)
{
return tuple_view(vals);
}
static tuple_view from(vals_t&& vals)
{
return tuple_view(std::move(vals));
}
tuple_view(const vals_t& vals, std::vector<size_t>&& mappings)
: m_vals(new detail::decorated_tuple<ElementTypes...>(vals, mappings))
{
......@@ -70,6 +80,14 @@ class tuple_view
private:
explicit tuple_view(const vals_t& vals) : m_vals(vals)
{
}
explicit tuple_view(vals_t&& vals) : m_vals(std::move(vals))
{
}
vals_t m_vals;
element_types m_types;
......
......@@ -28,6 +28,12 @@ void actor_proxy::forward_message(const process_information_ptr& piptr,
detail::mailman_queue().push_back(new detail::mailman_job(piptr, msg));
}
void actor_proxy::enqueue(any_tuple&& msg)
{
any_tuple tmp(std::move(msg));
enqueue(tmp);
}
void actor_proxy::enqueue(const any_tuple& msg)
{
if (msg.size() > 0 && msg.utype_info_at(0) == typeid(atom_value))
......
......@@ -43,11 +43,21 @@ throw_on_exit_result throw_on_exit(const any_tuple& msg)
namespace cppa { namespace detail {
blocking_message_queue_impl::queue_node::queue_node(const any_tuple& from)
: next(nullptr), msg(from)
blocking_message_queue_impl::queue_node::queue_node(any_tuple&& content)
: next(nullptr), msg(std::move(content))
{
}
blocking_message_queue_impl::queue_node::queue_node(const any_tuple& content)
: next(nullptr), msg(content)
{
}
void blocking_message_queue_impl::enqueue(any_tuple&& msg)
{
m_queue.push_back(new queue_node(std::move(msg)));
}
void blocking_message_queue_impl::enqueue(const any_tuple& msg)
{
m_queue.push_back(new queue_node(msg));
......
#include <set>
#include <stdexcept>
#include "cppa/any_tuple.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......@@ -44,6 +45,12 @@ class local_group : public group
}
}
virtual void enqueue(any_tuple&& msg)
{
any_tuple tmp(std::move(msg));
enqueue(tmp);
}
virtual group::subscription subscribe(const channel_ptr& who)
{
group::subscription result;
......
......@@ -29,6 +29,11 @@ boost::thread_specific_ptr<cppa::local_actor> t_this_context(cleanup_fun);
namespace cppa {
void local_actor::enqueue(any_tuple&& msg)
{
mailbox().enqueue(std::move(msg));
}
void local_actor::enqueue(const any_tuple& msg)
{
mailbox().enqueue(msg);
......
......@@ -29,8 +29,13 @@ typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard;
namespace cppa { namespace detail {
yielding_message_queue_impl::queue_node::queue_node(const any_tuple& from)
: next(0), msg(from)
yielding_message_queue_impl::queue_node::queue_node(any_tuple&& content)
: next(0), msg(std::move(content))
{
}
yielding_message_queue_impl::queue_node::queue_node(const any_tuple& content)
: next(0), msg(content)
{
}
......@@ -79,9 +84,9 @@ yielding_message_queue_impl::filter_msg(const any_tuple& msg)
return ordinary_message;
}
void yielding_message_queue_impl::enqueue(const any_tuple& msg)
void yielding_message_queue_impl::enqueue_node(queue_node* node)
{
if (m_queue._push_back(new queue_node(msg)))
if (m_queue._push_back(node))
{
for (;;)
{
......@@ -109,6 +114,16 @@ void yielding_message_queue_impl::enqueue(const any_tuple& msg)
}
}
void yielding_message_queue_impl::enqueue(any_tuple&& msg)
{
enqueue_node(new queue_node(std::move(msg)));
}
void yielding_message_queue_impl::enqueue(const any_tuple& msg)
{
enqueue_node(new queue_node(msg));
}
void yielding_message_queue_impl::yield_until_not_empty()
{
while (m_queue.empty())
......
#include <iostream>
#include "ping_pong.hpp"
#include "cppa/cppa.hpp"
......@@ -5,43 +7,58 @@
namespace { int s_pongs = 0; }
using std::cout;
using std::endl;
using namespace cppa;
void pong(actor_ptr ping_actor)
void pong(actor_ptr ping_actor);
void ping()
{
link(ping_actor);
// kickoff
send(ping_actor, atom("Pong"), static_cast<std::int32_t>(0));
// invoke rules
s_pongs = 0;
actor_ptr pong_actor;
// invoke rule
receive_loop
(
on(atom("Ping"), val<actor_ptr>(), std::int32_t(9)) >> []()
//on<atom("Ping"), std::int32_t>(9) >> []()
on<atom("Pong"), std::int32_t>() >> [&](std::int32_t value)
{
// terminate with non-normal exit reason
// to force ping actor to quit
quit(exit_reason::user_defined);
++s_pongs;
auto msg = tuple_cast<atom_value, std::int32_t>(std::move(last_received()));
get_ref<0>(msg) = atom("Ping");
get_ref<1>(msg) = value + 1;
pong_actor->enqueue(std::move(msg));
//send(pong_actor, atom("Ping"), value + 1);
},
on<atom("Ping"), actor_ptr, std::int32_t>() >> [](actor_ptr teammate,
std::int32_t value)
//on<atom("Ping"), std::int32_t>() >> [](int value)
on<atom("Hello"), actor_ptr>() >> [&](actor_ptr sender)
{
send(teammate, atom("Ping"), self(), value + 1);
pong_actor = sender;
}
);
}
void ping()
void pong(actor_ptr ping_actor)
{
s_pongs = 0;
// invoke rule
link(ping_actor);
// tell ping who we are
send(ping_actor, atom("Hello"), self());
// kickoff
send(ping_actor, atom("Pong"), static_cast<std::int32_t>(0));
// invoke rules
receive_loop
(
on<atom("Pong"), actor_ptr, std::int32_t>() >> [](actor_ptr teammate,
std::int32_t value)
on(atom("Ping"), std::int32_t(9)) >> []()
{
++s_pongs;
send(teammate, atom("Ping"), self(), value + 1);
// terminate with non-normal exit reason
// to force ping actor to quit
quit(exit_reason::user_defined);
},
on<atom("Ping"), std::int32_t>() >> [&](std::int32_t value)
{
auto msg = tuple_cast<atom_value, std::int32_t>(std::move(last_received()));
get_ref<0>(msg) = atom("Pong");
get_ref<1>(msg) = value + 1;
ping_actor->enqueue(std::move(msg));
//send(ping_actor, atom("Pong"), value + 1);
}
);
}
......
......@@ -70,7 +70,6 @@ void testee3(actor_ptr parent)
size_t test__spawn()
{
return 0;
CPPA_TEST(test__spawn);
auto report_unexpected = [&]()
{
......@@ -81,12 +80,12 @@ size_t test__spawn()
auto pong_actor = spawn(pong, spawn(ping));
monitor(pong_actor);
link(pong_actor);
monitor(spawn(testee2, spawn(testee1)));
// monitor(spawn(testee2, spawn(testee1)));
int i = 0;
int flags = 0;
future_send(self(), std::chrono::seconds(1), atom("FooBar"));
// wait for :Down and :Exit messages of pong
receive_while([&i]() { return ++i <= 4; })
receive_while([&i]() { return ++i <= 3 /*4*/; })
(
on<atom(":Exit"), actor_ptr, std::uint32_t>() >> [&](const actor_ptr& who,
std::uint32_t reason)
......@@ -103,14 +102,10 @@ size_t test__spawn()
{
flags |= 0x02;
}
else
{
flags |= 0x04;
}
},
on<atom("FooBar")>() >> [&]()
{
flags |= 0x08;
flags |= 0x04;
},
others() >> [&]()
{
......@@ -126,7 +121,7 @@ size_t test__spawn()
);
// wait for termination of all spawned actors
await_all_others_done();
CPPA_CHECK_EQUAL(flags, 0x0F);
CPPA_CHECK_EQUAL(flags, 0x07);
// mailbox has to be empty
any_tuple msg;
while (try_receive(msg))
......@@ -135,6 +130,7 @@ size_t test__spawn()
}
// verify pong messages
CPPA_CHECK_EQUAL(pongs(), 5);
/*
spawn(testee3, self());
i = 0;
// testee3 sends 5 { "Push", int } messages in a 50 milliseconds interval;
......@@ -154,5 +150,6 @@ size_t test__spawn()
}
);
await_all_others_done();
*/
return CPPA_TEST_RESULT;
}
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