Commit c05a8c82 authored by neverlord's avatar neverlord

maintenance

parent 82853dc5
...@@ -34,7 +34,6 @@ ...@@ -34,7 +34,6 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include <list> #include <list>
#include <mutex>
#include <atomic> #include <atomic>
#include <vector> #include <vector>
#include <memory> #include <memory>
...@@ -45,6 +44,7 @@ ...@@ -45,6 +44,7 @@
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/attachable.hpp" #include "cppa/attachable.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/detail/thread.hpp"
namespace cppa { namespace cppa {
...@@ -52,15 +52,16 @@ namespace cppa { ...@@ -52,15 +52,16 @@ namespace cppa {
template<class Base> template<class Base>
class abstract_actor : public Base class abstract_actor : public Base
{ {
typedef detail::lock_guard<detail::mutex> guard_type;
typedef std::lock_guard<std::mutex> guard_type; //typedef std::lock_guard<std::mutex> guard_type;
typedef std::unique_ptr<attachable> attachable_ptr; typedef std::unique_ptr<attachable> attachable_ptr;
// true if the associated thread has finished execution // true if the associated thread has finished execution
std::atomic<std::uint32_t> m_exit_reason; std::atomic<std::uint32_t> m_exit_reason;
// guards access to m_exited, m_subscriptions and m_links // guards access to m_exited, m_subscriptions and m_links
std::mutex m_mtx; //std::mutex m_mtx;
detail::mutex m_mtx;
// manages actor links // manages actor links
std::list<actor_ptr> m_links; std::list<actor_ptr> m_links;
...@@ -78,13 +79,11 @@ class abstract_actor : public Base ...@@ -78,13 +79,11 @@ class abstract_actor : public Base
queue_node(actor* from, any_tuple&& content) queue_node(actor* from, any_tuple&& content)
: next(nullptr), sender(from), msg(std::move(content)) : next(nullptr), sender(from), msg(std::move(content))
{ {
} }
queue_node(actor* from, any_tuple const& content) queue_node(actor* from, any_tuple const& content)
: next(nullptr), sender(from), msg(content) : next(nullptr), sender(from), msg(content)
{ {
} }
}; };
...@@ -130,8 +129,6 @@ class abstract_actor : public Base ...@@ -130,8 +129,6 @@ class abstract_actor : public Base
protected: protected:
//abstract_actor() : Base() { }
template<typename... Args> template<typename... Args>
abstract_actor(Args&&... args) : Base(std::forward<Args>(args)...) abstract_actor(Args&&... args) : Base(std::forward<Args>(args)...)
, m_exit_reason(exit_reason::not_exited) , m_exit_reason(exit_reason::not_exited)
...@@ -145,7 +142,7 @@ class abstract_actor : public Base ...@@ -145,7 +142,7 @@ class abstract_actor : public Base
decltype(m_attachables) mattachables; decltype(m_attachables) mattachables;
// lifetime scope of guard // lifetime scope of guard
{ {
std::lock_guard<std::mutex> guard(m_mtx); guard_type guard(m_mtx);
if (m_exit_reason != exit_reason::not_exited) if (m_exit_reason != exit_reason::not_exited)
{ {
// already exited // already exited
...@@ -174,18 +171,25 @@ class abstract_actor : public Base ...@@ -174,18 +171,25 @@ class abstract_actor : public Base
bool link_to_impl(intrusive_ptr<actor>& other) bool link_to_impl(intrusive_ptr<actor>& other)
{ {
guard_type guard(m_mtx); if (other && other != this)
if (other && !exited() && other->establish_backlink(this))
{ {
m_links.push_back(other); guard_type guard(m_mtx);
return true; if (exited())
{
other->enqueue(this, atom(":Exit"), m_exit_reason.load());
}
else if (other->establish_backlink(this))
{
m_links.push_back(other);
return true;
}
} }
return false; return false;
} }
bool unlink_from_impl (intrusive_ptr<actor>& other) bool unlink_from_impl(intrusive_ptr<actor>& other)
{ {
std::lock_guard<std::mutex> guard(m_mtx); guard_type guard(m_mtx);
if (other && !exited() && other->remove_backlink(this)) if (other && !exited() && other->remove_backlink(this))
{ {
return erase_all(m_links, other) > 0; return erase_all(m_links, other) > 0;
...@@ -227,15 +231,17 @@ class abstract_actor : public Base ...@@ -227,15 +231,17 @@ class abstract_actor : public Base
// lifetime scope of guard // lifetime scope of guard
{ {
guard_type guard(m_mtx); guard_type guard(m_mtx);
for (auto i = m_attachables.begin(); i != m_attachables.end(); ++i) auto end = m_attachables.end();
auto i = std::find_if(m_attachables.begin(),
end,
[&](attachable_ptr& ptr)
{
return ptr->matches(what);
});
if (i != end)
{ {
if ((*i)->matches(what)) uptr = std::move(*i);
{ m_attachables.erase(i);
uptr = std::move(*i);
m_attachables.erase(i);
// exit loop (and release lock)
break;
}
} }
} }
// uptr will be destroyed here, without locked mutex // uptr will be destroyed here, without locked mutex
...@@ -255,34 +261,30 @@ class abstract_actor : public Base ...@@ -255,34 +261,30 @@ class abstract_actor : public Base
{ {
if (other && other != this) if (other && other != this)
{ {
std::lock_guard<std::mutex> guard(m_mtx); guard_type guard(m_mtx);
return erase_all(m_links, other) > 0;//m_links.erase(other) > 0; return erase_all(m_links, other) > 0;
} }
return false; return false;
} }
bool establish_backlink(intrusive_ptr<actor>& other) /*override*/ bool establish_backlink(intrusive_ptr<actor>& other) /*override*/
{ {
bool result = false;
std::uint32_t reason = exit_reason::not_exited; std::uint32_t reason = exit_reason::not_exited;
if (other && other != this) if (other && other != this)
{ {
// lifetime scope of guard guard_type guard(m_mtx);
reason = m_exit_reason.load();
if (reason == exit_reason::not_exited)
{ {
std::lock_guard<std::mutex> guard(m_mtx); return unique_insert(m_links, other);
reason = m_exit_reason.load();
if (reason == exit_reason::not_exited)
{
result = unique_insert(m_links, other);
//result = m_links.insert(other).second;
}
} }
} }
// send exit message without lock
if (reason != exit_reason::not_exited) if (reason != exit_reason::not_exited)
{ {
other->enqueue(this, make_tuple(atom(":Exit"), reason)); other->enqueue(this, make_tuple(atom(":Exit"), reason));
} }
return result; return false;
} }
}; };
......
...@@ -167,9 +167,9 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor ...@@ -167,9 +167,9 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor
template<typename... Args> template<typename... Args>
void receive(Args&&...) void receive(Args&&...)
{ {
static_assert(sizeof...(Args) < 0, static_assert((sizeof...(Args) + 1) < 1,
"You shall not use receive in an event-based actor. " "You shall not use receive in an event-based actor. "
"Use become()/unbecome() instead."); "Use become() instead.");
} }
template<typename... Args> template<typename... Args>
......
...@@ -35,9 +35,8 @@ namespace cppa { ...@@ -35,9 +35,8 @@ namespace cppa {
enum scheduling_hint enum scheduling_hint
{ {
scheduled, scheduled,
detached, detached
as_task
}; };
} // namespace cppa } // namespace cppa
......
...@@ -101,32 +101,29 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) ...@@ -101,32 +101,29 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
{ {
// setup & local variables // setup & local variables
self.set(m_self.get()); self.set(m_self.get());
//set_self(m_self.get());
auto& queue = m_self->mailbox(); auto& queue = m_self->mailbox();
//typedef std::pair<cppa::actor_ptr, decltype(queue.pop())> future_msg;
std::multimap<decltype(detail::now()), decltype(queue.pop())> messages; std::multimap<decltype(detail::now()), decltype(queue.pop())> messages;
decltype(queue.pop()) msg_ptr = nullptr; decltype(queue.pop()) msg_ptr = nullptr;
decltype(detail::now()) now; decltype(detail::now()) now;
bool done = false; bool done = false;
// message handling rules // message handling rules
auto rules = auto handle_msg =
( (
on<util::duration,actor_ptr,anything>() >> [&](const util::duration& d, on<util::duration,actor_ptr,anything>() >> [&](const util::duration& d,
const actor_ptr&) const actor_ptr&)
// on<util::duration,anything>() >> [&](const util::duration& d)
{ {
//any_tuple msg = msg_ptr->msg.tail();
// calculate timeout // calculate timeout
auto timeout = detail::now(); auto timeout = detail::now();
timeout += d; timeout += d;
messages.insert(std::make_pair(std::move(timeout), messages.insert(std::make_pair(std::move(timeout),
std::move(msg_ptr))); std::move(msg_ptr)));
// do not delete this msg_ptr (now)
msg_ptr = nullptr;
}, },
on<atom(":_DIE")>() >> [&]() on<atom(":_DIE")>() >> [&]()
{ {
done = true; done = true;
}, }
others() >> []() { }
); );
// loop // loop
while (!done) while (!done)
...@@ -145,7 +142,9 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) ...@@ -145,7 +142,9 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
while (it != messages.end() && (it->first) <= now) while (it != messages.end() && (it->first) <= now)
{ {
auto ptr = it->second; auto ptr = it->second;
auto whom = const_cast<actor_ptr*>(reinterpret_cast<const actor_ptr*>(ptr->msg.at(1))); auto whom = const_cast<actor_ptr*>(
reinterpret_cast<actor_ptr const*>(
ptr->msg.at(1)));
if (*whom) if (*whom)
{ {
auto msg = ptr->msg.tail(2); auto msg = ptr->msg.tail(2);
...@@ -162,9 +161,8 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) ...@@ -162,9 +161,8 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
} }
} }
} }
rules(msg_ptr->msg); handle_msg(msg_ptr->msg);
//delete msg_ptr; delete msg_ptr;
msg_ptr = nullptr;
} }
} }
......
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