Commit 3d38389e authored by neverlord's avatar neverlord

scheduled_actor::execute

parent 03fde419
...@@ -9,7 +9,9 @@ ...@@ -9,7 +9,9 @@
#include <signal.h> #include <signal.h>
#include <stddef.h> #include <stddef.h>
#include "cppa_fibre.h" static ucontext_t ctx[2];
__thread int m_count = 0;
void coroutine() void coroutine()
{ {
...@@ -41,7 +43,7 @@ int main(int argc, char** argv) ...@@ -41,7 +43,7 @@ int main(int argc, char** argv)
ctx[1].uc_link = &ctx[0]; ctx[1].uc_link = &ctx[0];
makecontext(&ctx[1], coroutine, 0); makecontext(&ctx[1], coroutine, 0);
for (i = 0; i < 11; ++i) for (i = 1; i < 11; ++i)
{ {
printf("i = %i\n", i); printf("i = %i\n", i);
swapcontext(&ctx[0], &ctx[1]); swapcontext(&ctx[0], &ctx[1]);
......
...@@ -88,32 +88,22 @@ class scheduled_actor : public abstract_actor<local_actor> ...@@ -88,32 +88,22 @@ class scheduled_actor : public abstract_actor<local_actor>
// otherwise false // otherwise false
template<typename StillReadyHandler, typename DoneHandler> template<typename StillReadyHandler, typename DoneHandler>
static void execute(scheduled_actor* what, static void execute(scheduled_actor* what,
util::fiber& from, util::fiber* from,
StillReadyHandler&& still_ready_handler, StillReadyHandler& still_ready_handler,
DoneHandler&& done_handler); DoneHandler& done_handler);
template<typename DoneHandler>
inline static void execute(scheduled_actor* what,
util::fiber& from,
DoneHandler&& done_handler)
{
execute(what, from,
[]() -> bool { return true; },
std::forward<DoneHandler>(done_handler));
}
}; };
template<typename StillReadyHandler, typename DoneHandler> template<typename StillReadyHandler, typename DoneHandler>
void scheduled_actor::execute(scheduled_actor* what, void scheduled_actor::execute(scheduled_actor* what,
util::fiber& from, util::fiber* from,
StillReadyHandler&& still_ready_handler, StillReadyHandler& still_ready_handler,
DoneHandler&& done_handler) DoneHandler& done_handler)
{ {
set_self(what); set_self(what);
for (;;) for (;;)
{ {
call(&(what->m_fiber), &from); call(&(what->m_fiber), from);
switch (yielded_state()) switch (yielded_state())
{ {
case yield_state::done: case yield_state::done:
...@@ -152,7 +142,7 @@ void scheduled_actor::execute(scheduled_actor* what, ...@@ -152,7 +142,7 @@ void scheduled_actor::execute(scheduled_actor* what,
default: default:
{ {
// illegal state // illegal state
exit(7); exit(8);
} }
} }
} }
......
...@@ -28,19 +28,22 @@ namespace cppa { namespace detail { ...@@ -28,19 +28,22 @@ namespace cppa { namespace detail {
void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy) void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy)
{ {
cppa::util::fiber fself; cppa::util::fiber fself;
scheduled_actor* job = nullptr;
auto still_ready_cb = []() { return true; };
auto done_cb = [&]()
{
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
};
for (;;) for (;;)
{ {
scheduled_actor* job = jq->pop(); job = jq->pop();
if (job == dummy) if (job == dummy)
{ {
return; return;
} }
scheduled_actor::execute(job, fself, [&]() scheduled_actor::execute(job, &fself, still_ready_cb, done_cb);
{
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
});
} }
} }
......
...@@ -58,6 +58,25 @@ struct thread_pool_scheduler::worker ...@@ -58,6 +58,25 @@ struct thread_pool_scheduler::worker
m_supervisor_queue->push_back(this); m_supervisor_queue->push_back(this);
// loop // loop
util::fiber fself; util::fiber fself;
auto tout = now();
bool reschedule = false;
auto still_ready_cb = [&]() -> bool
{
if (tout >= now())
{
reschedule = true;
return false;
}
return true;
};
scheduled_actor* job;
auto done_cb = [&]()
{
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
job = nullptr;
};
for (;;) for (;;)
{ {
// lifetime scope of guard (wait for new job) // lifetime scope of guard (wait for new job)
...@@ -71,106 +90,15 @@ struct thread_pool_scheduler::worker ...@@ -71,106 +90,15 @@ struct thread_pool_scheduler::worker
} }
auto job = const_cast<scheduled_actor*>(m_job); auto job = const_cast<scheduled_actor*>(m_job);
// run actor up to 300ms // run actor up to 300ms
bool reschedule = false; reschedule = false;
auto tout = now(); tout = now();
tout += std::chrono::milliseconds(300); tout += std::chrono::milliseconds(300);
set_self(job); scheduled_actor::execute(job, &fself, still_ready_cb, done_cb);
CPPA_MEMORY_BARRIER();
bool job_done = false;
do
{
cout << "switch context\n";
call(job->fiber_ptr(), &fself);
CPPA_MEMORY_BARRIER();
switch (yielded_state())
{
case yield_state::done:
case yield_state::killed:
{
cout << "done | killed\n";
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
job = nullptr;
job_done = true;
break;
}
case yield_state::ready:
{
cout << "ready\n";
if (tout >= now())
{
reschedule = true;
job_done = true;
}
break;
}
case yield_state::blocked:
{
cout << "blocked\n";
cout << "job addr: " << std::hex << job << endl;
switch (job->compare_exchange_state(about_to_block,
blocked))
{
case ready:
{
cout << "blocked -> ready\n";
if (tout >= now())
{
reschedule = true;
job_done = true;
}
break;
}
case blocked:
{
cout << "blocked -> blocked\n";
// wait until someone re-schedules that actor
break;
}
default:
{
cout << "blocked -> illegal state\n";
// illegal state
exit(7);
}
}
break;
}
default:
{
cout << "illegal state\n";
// illegal state
exit(8);
}
}
}
while (job_done == false);
/*
scheduled_actor::execute(m_job,
fself,
[&]() -> bool
{
if (tout >= now())
{
reschedule = true;
return false;
}
return true;
},
[m_job]()
{
if (!m_job->deref()) delete m_job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
});
*/
if (reschedule && job) if (reschedule && job)
{ {
m_job_queue->push_back(job); m_job_queue->push_back(job);
} }
m_job = nullptr; m_job = nullptr;
cout << "wait for next job\n";
CPPA_MEMORY_BARRIER(); CPPA_MEMORY_BARRIER();
m_supervisor_queue->push_back(this); m_supervisor_queue->push_back(this);
} }
......
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