Commit a7038dc9 authored by neverlord's avatar neverlord

pending_enqueue

parent c05472d2
......@@ -591,7 +591,7 @@ template<class C, typename Arg0, typename... Args>
void send(intrusive_ptr<C>& whom, Arg0 const& arg0, Args const&... args)
{
static_assert(std::is_base_of<channel, C>::value, "C is not a channel");
if (whom) whom->enqueue(self, make_cow_tuple(arg0, args...));
if (whom) self->send_message(whom.get(), make_cow_tuple(arg0, args...));
}
template<class C, typename Arg0, typename... Args>
......@@ -599,13 +599,14 @@ void send(intrusive_ptr<C>&& whom, Arg0 const& arg0, Args const&... args)
{
static_assert(std::is_base_of<channel, C>::value, "C is not a channel");
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(self, make_cow_tuple(arg0, args...));
send(tmp, arg0, args...);
}
// matches "send(this, ...)" in event-based actors
template<typename Arg0, typename... Args>
void send(local_actor* whom, Arg0 const& arg0, Args const&... args)
inline void send(local_actor* whom, Arg0 const& arg0, Args const&... args)
{
CPPA_REQUIRE(whom != nullptr);
whom->enqueue(whom, make_cow_tuple(arg0, args...));
}
......@@ -624,7 +625,7 @@ typename std::enable_if<
>::type
operator<<(intrusive_ptr<C>& whom, any_tuple const& what)
{
if (whom) whom->enqueue(self, what);
if (whom) self->send_message(whom.get(), what);
return whom;
}
......@@ -636,7 +637,7 @@ typename std::enable_if<
operator<<(intrusive_ptr<C>&& whom, any_tuple const& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(self, what);
tmp << what;
return std::move(tmp);
}
......@@ -647,7 +648,7 @@ typename std::enable_if<
>::type
operator<<(intrusive_ptr<C>& whom, any_tuple&& what)
{
if (whom) whom->enqueue(self, std::move(what));
if (whom) self->send_message(whom.get(), std::move(what));
return whom;
}
......@@ -659,7 +660,7 @@ typename std::enable_if<
operator<<(intrusive_ptr<C>&& whom, any_tuple&& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(self, std::move(what));
tmp << std::move(what);
return std::move(tmp);
}
......
......@@ -124,6 +124,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
static constexpr int ready = 0x00;
static constexpr int done = 0x01;
static constexpr int blocked = 0x02;
static constexpr int pending = 0x03;
static constexpr int about_to_block = 0x04;
abstract_scheduled_actor(int state = done)
......@@ -133,6 +134,11 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
{
}
bool pending_enqueue(actor* sender, any_tuple msg)
{
return enqueue_node(super::fetch_node(sender, std::move(msg)), pending);
}
void quit(std::uint32_t reason)
{
this->cleanup(reason);
......@@ -160,7 +166,8 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
private:
void enqueue_node(typename super::mailbox_element* node)
bool enqueue_node(typename super::mailbox_element* node,
int target_state = ready)
{
if (this->m_mailbox._push_back(node))
{
......@@ -171,26 +178,27 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
{
case blocked:
{
if (m_state.compare_exchange_weak(state, ready))
if (m_state.compare_exchange_weak(state, target_state))
{
CPPA_REQUIRE(this->m_scheduler != nullptr);
this->m_scheduler->enqueue(this);
return;
return true;
}
break;
}
case about_to_block:
{
if (m_state.compare_exchange_weak(state, ready))
if (m_state.compare_exchange_weak(state, target_state))
{
return;
return true;
}
break;
}
default: return;
default: return false;
}
}
}
return false;
}
};
......
......@@ -52,12 +52,14 @@ class local_actor : public actor
protected:
bool m_trap_exit;
bool m_is_scheduled;
actor_ptr m_pending;
actor_ptr m_last_sender;
any_tuple m_last_dequeued;
public:
local_actor();
local_actor(bool is_scheduled = false);
/**
* @brief Finishes execution of this actor.
......@@ -93,6 +95,12 @@ class local_actor : public actor
inline actor_ptr& last_sender();
inline actor_ptr& pending_actor();
inline void send_message(channel* whom, any_tuple what);
inline void send_message(actor* whom, any_tuple what);
};
inline bool local_actor::trap_exit() const
......@@ -115,6 +123,31 @@ inline actor_ptr& local_actor::last_sender()
return m_last_sender;
}
inline actor_ptr& local_actor::pending_actor()
{
return m_pending;
}
inline void local_actor::send_message(actor* whom, any_tuple what)
{
if (m_is_scheduled && !m_pending)
{
if (whom->pending_enqueue(this, std::move(what)))
{
m_pending = whom;
}
}
else
{
whom->enqueue(this, std::move(what));
}
}
inline void local_actor::send_message(channel* whom, any_tuple what)
{
whom->enqueue(this, std::move(what));
}
typedef intrusive_ptr<local_actor> local_actor_ptr;
} // namespace cppa
......
......@@ -36,6 +36,7 @@
#include <stdexcept>
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......
......@@ -32,7 +32,7 @@
namespace cppa {
local_actor::local_actor() : m_trap_exit(false)
local_actor::local_actor(bool sflag) : m_trap_exit(false), m_is_scheduled(sflag)
{
}
......
......@@ -137,9 +137,15 @@ struct thread_pool_scheduler::worker
struct handler : scheduler::callback
{
scheduled_actor* job;
handler() : job(nullptr) { }
scheduled_actor* pending_job;
handler() : job(nullptr), pending_job(nullptr) { }
void exec_done()
{
if (job->pending_actor())
{
pending_job = static_cast<scheduled_actor*>(job->pending_actor().get());
job->pending_actor().reset();
}
if (!job->deref()) delete job;
dec_actor_count();
job = nullptr;
......@@ -165,7 +171,28 @@ struct thread_pool_scheduler::worker
}
else
{
h.job->resume(&fself, &h);
do
{
h.job->resume(&fself, &h);
if (h.job)
{
if (h.job->pending_actor())
{
h.pending_job = static_cast<scheduled_actor*>(h.job->pending_actor().get());
h.job->pending_actor().reset();
}
else
{
h.job = nullptr;
}
}
if (h.pending_job)
{
h.job = h.pending_job;
h.pending_job = nullptr;
}
}
while (h.job);
}
}
}
......
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