Commit 209a5006 authored by Neverlord's avatar Neverlord

fixed memory handling of scheduled actors

parent c338a2e8
...@@ -70,6 +70,14 @@ class context_switching_resume { ...@@ -70,6 +70,14 @@ class context_switching_resume {
, m_cs_thread(context_switching_resume::trampoline, , m_cs_thread(context_switching_resume::trampoline,
static_cast<blocking_actor*>(this)) { } static_cast<blocking_actor*>(this)) { }
void attach_to_scheduler() override {
this->ref();
}
void detach_from_scheduler() override {
this->deref();
}
resumable::resume_result resume(detail::cs_thread* from, resumable::resume_result resume(detail::cs_thread* from,
execution_unit* host) override { execution_unit* host) override {
CPPA_REQUIRE(from != nullptr); CPPA_REQUIRE(from != nullptr);
......
...@@ -83,6 +83,8 @@ class cooperative_scheduling { ...@@ -83,6 +83,8 @@ class cooperative_scheduling {
template<class Actor> template<class Actor>
inline void launch(Actor* self) { inline void launch(Actor* self) {
// detached in scheduler::worker::run
self->attach_to_scheduler();
get_scheduling_coordinator()->enqueue(self); get_scheduling_coordinator()->enqueue(self);
} }
......
...@@ -64,6 +64,14 @@ class event_based_resume { ...@@ -64,6 +64,14 @@ class event_based_resume {
return static_cast<Derived*>(this); return static_cast<Derived*>(this);
} }
void attach_to_scheduler() override {
this->ref();
}
void detach_from_scheduler() override {
this->deref();
}
resumable::resume_result resume(detail::cs_thread*, resumable::resume_result resume(detail::cs_thread*,
execution_unit* host) override { execution_unit* host) override {
auto d = dptr(); auto d = dptr();
......
...@@ -31,6 +31,14 @@ class no_resume { ...@@ -31,6 +31,14 @@ class no_resume {
: Base(std::forward<Ts>(args)...) : Base(std::forward<Ts>(args)...)
, m_hidden(true) { } , m_hidden(true) { }
inline void attach_to_scheduler() {
this->ref();
}
inline void detach_from_scheduler() {
this->deref();
}
inline resumable::resume_result resume(detail::cs_thread*, inline resumable::resume_result resume(detail::cs_thread*,
execution_unit*) { execution_unit*) {
auto done_cb = [=](std::uint32_t reason) { auto done_cb = [=](std::uint32_t reason) {
......
...@@ -37,6 +37,10 @@ class execution_unit; ...@@ -37,6 +37,10 @@ class execution_unit;
namespace detail { struct cs_thread; } namespace detail { struct cs_thread; }
/**
* @brief A cooperatively executed task managed by one or more instances of
* {@link execution_unit}.
*/
class resumable { class resumable {
public: public:
...@@ -46,14 +50,26 @@ class resumable { ...@@ -46,14 +50,26 @@ class resumable {
done done
}; };
// intrusive next pointer needed to use
// 'resumable' with 'single_reader_queue'
resumable* next;
resumable(); resumable();
virtual ~resumable(); virtual ~resumable();
/**
* @brief Initializes this object, e.g., by increasing the
* the reference count.
*/
virtual void attach_to_scheduler() = 0;
/**
* @brief Uninitializes this object, e.g., by decrementing the
* the reference count.
*/
virtual void detach_from_scheduler() = 0;
/**
* @brief Resume any pending computation until it is either finished
* or needs to be re-scheduled later.
*/
virtual resume_result resume(detail::cs_thread*, execution_unit*) = 0; virtual resume_result resume(detail::cs_thread*, execution_unit*) = 0;
protected: protected:
......
...@@ -32,7 +32,7 @@ ...@@ -32,7 +32,7 @@
namespace cppa { namespace cppa {
resumable::resumable() : next(nullptr), m_hidden(true) { } resumable::resumable() : m_hidden(true) { }
resumable::~resumable() { } resumable::~resumable() { }
......
...@@ -235,6 +235,10 @@ class coordinator::shutdown_helper : public resumable { ...@@ -235,6 +235,10 @@ class coordinator::shutdown_helper : public resumable {
public: public:
void attach_to_scheduler() override { }
void detach_from_scheduler() override { }
resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) { resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) {
auto w = dynamic_cast<worker*>(ptr); auto w = dynamic_cast<worker*>(ptr);
CPPA_REQUIRE(w != nullptr); CPPA_REQUIRE(w != nullptr);
...@@ -260,12 +264,13 @@ void coordinator::initialize() { ...@@ -260,12 +264,13 @@ void coordinator::initialize() {
ptr->act(); ptr->act();
}}; }};
m_printer_thread = std::thread{printer_loop, m_printer.get()}; m_printer_thread = std::thread{printer_loop, m_printer.get()};
// launch workers // create workers
size_t hc = std::thread::hardware_concurrency(); size_t hc = std::thread::hardware_concurrency();
for (size_t i = 0; i < hc; ++i) { for (size_t i = 0; i < hc; ++i) {
m_workers.emplace_back(new worker(i, this)); m_workers.emplace_back(new worker(i, this));
m_workers.back()->start();
} }
// start all workers
for (auto& w : m_workers) w->start();
} }
void coordinator::destroy() { void coordinator::destroy() {
...@@ -299,7 +304,9 @@ void coordinator::destroy() { ...@@ -299,7 +304,9 @@ void coordinator::destroy() {
delete this; delete this;
} }
coordinator::coordinator() : m_timer(new timer_actor), m_printer(true) { coordinator::coordinator()
: m_timer(new timer_actor), m_printer(true)
, m_next_worker(0) {
// NOP // NOP
} }
...@@ -386,7 +393,10 @@ void worker::run() { ...@@ -386,7 +393,10 @@ void worker::run() {
while (m_running) { while (m_running) {
local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll(); local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll();
CPPA_LOG_DEBUG("dequeued new job"); CPPA_LOG_DEBUG("dequeued new job");
job->resume(&fself, this); if (job->resume(&fself, this) == resumable::done) {
// was attached in policy::cooperative_scheduling::launch
job->detach_from_scheduler();
}
job = nullptr; job = 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