Commit 0d233f48 authored by neverlord's avatar neverlord

towards a new yield interface

parent 2c4a331f
...@@ -75,7 +75,8 @@ cppatest_SOURCES = \ ...@@ -75,7 +75,8 @@ cppatest_SOURCES = \
../unit_testing/test__spawn.cpp \ ../unit_testing/test__spawn.cpp \
../unit_testing/test__tuple.cpp \ ../unit_testing/test__tuple.cpp \
../unit_testing/test__type_list.cpp \ ../unit_testing/test__type_list.cpp \
../unit_testing/test__uniform_type.cpp ../unit_testing/test__uniform_type.cpp \
../unit_testing/test__yield_interface.cpp
AM_CPPFLAGS = -I../ AM_CPPFLAGS = -I../
cppatest_CXXFLAGS = --std=c++0x -pedantic -Wall -Wextra cppatest_CXXFLAGS = --std=c++0x -pedantic -Wall -Wextra
......
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif
#include <stdio.h>
#include <ucontext.h>
#include <sys/mman.h>
#include <signal.h>
#include <stddef.h>
#include "cppa_fibre.h"
static ucontext_t ctx[2];
__thread int m_count = 0;
struct cppa_fibre
{
ucontext_t m_context;
};
void cppa_fibre_ctor(cppa_fibre* instance);
void cppa_fibre_ctor(cppa_fibre* instance, void (*fun)());
void cppa_fibre_dtor(cppa_fibre* instance);
void cppa_fibre_switch(cppa_fibre* from, cppa_fibre* to);
void cppa_fibre_yield(int value);
int cppa_fibre_yielded_value();
typedef struct fibre_wrapper* fibre_ptr;
fibre_ptr get_fibre();
void coroutine()
{
for (;;)
{
++m_count;
printf("m_count = %i\n", m_count);
swapcontext(&ctx[1], &ctx[0]);
}
}
int main(int argc, char** argv)
{
int i;
void* coroutine_stack = mmap(0,
SIGSTKSZ,
PROT_EXEC | PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON,
-1,
0);
memset(&ctx[0], 0, sizeof(ucontext_t));
getcontext(&ctx[0]);
memset(&ctx[1], 0, sizeof(ucontext_t));
getcontext(&ctx[1]);
ctx[1].uc_stack.ss_sp = coroutine_stack;
ctx[1].uc_stack.ss_size = SIGSTKSZ;
ctx[1].uc_link = &ctx[0];
makecontext(&ctx[1], coroutine, 0);
for (i = 0; i < 11; ++i)
{
printf("i = %i\n", i);
swapcontext(&ctx[0], &ctx[1]);
}
munmap(coroutine_stack, SIGSTKSZ);
return 0;
}
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif
#include <stdio.h>
#include <ucontext.h>
#include <sys/mman.h>
#include <signal.h>
#include <stddef.h>
#include "cppa_fibre.h"
/*
struct cppa_fibre_struct
{
int m_initialized;
ucontext_t m_context;
void (*m_fun)();
void* m_init_arg;
};
*/
void cppa_fibre_ctor(cppa_fibre* instance)
{
instance->m_state = 0;
memset(&(instance->m_context), 0, sizeof(ucontext_t));
getcontext(&(instance->m_context));
instance->m_fun = 0;
instance->m_init_arg = 0;
}
void cppa_fibre_ctor2(cppa_fibre* instance, void (*fun)(), void* arg)
{
cppa_fibre_ctor(instance);
instance->m_state = 1;
instance->m_fun = fun;
instance->m_init_arg = arg;
}
void cppa_fibre_initialize(cppa_fibre* instance)
{
if (instance->m_state == 1)
{
instance->m_state = 2;
instance->m_context.uc_stack.ss_sp = mmap(0,
SIGSTKSZ,
PROT_EXEC | PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON,
-1,
0);
instance->m_context.uc_stack.ss_size = SIGSTKSZ;
makecontext(&(instance->m_context), instance->m_fun, 0);
}
}
void cppa_fibre_dtor(cppa_fibre* instance)
{
}
/*
* @brief Returns
*/
void* cppa_fibre_init_switch_arg();
void cppa_fibre_switch(cppa_fibre* from, cppa_fibre* to);
/*
* Switches back to the calling fibre.
*/
void cppa_fibre_yield(int value);
/*
* Gets the yielded value of the client fibre.
*/
int cppa_fibre_yielded_value();
#ifdef __cplusplus
}
#endif
#endif
#ifndef CPPA_FIBRE_H
#define CPPA_FIBRE_H
#ifdef __cplusplus
extern "C" {
#endif
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif
#include <stdio.h>
#include <ucontext.h>
#include <sys/mman.h>
#include <signal.h>
#include <stddef.h>
struct cppa_fibre_struct
{
// 0: *this* context
// 1: fibre with function to execute, no stack assigned yet
// 2: as 1 but with assigned stack
int m_state;
ucontext_t m_context;
void (*m_fun)();
void* m_init_arg;
};
typedef cppa_fibre_struct cppa_fibre;
void cppa_fibre_ctor(cppa_fibre* instance);
/*
* @brief Initializes the given fibre.
* @param instance Pointer to an uninitialized object.
* @param fun Function this fibre should execute.
* @param switch_arg This pointer is stored in a
* thread-local variable on first
* context switch to @p instance.
*/
void cppa_fibre_ctor2(cppa_fibre* instance,
void (*fun)(),
void* switch_arg);
void cppa_fibre_initialize(cppa_fibre* instance);
void cppa_fibre_dtor(cppa_fibre* instance);
/*
* @brief Returns
*/
void* cppa_fibre_init_switch_arg();
void cppa_fibre_switch(cppa_fibre* from, cppa_fibre* to);
/*
* Switches back to the calling fibre.
*/
void cppa_fibre_yield(int value);
/*
* Gets the yielded value of the client fibre.
*/
int cppa_fibre_yielded_value();
#ifdef __cplusplus
}
#endif
#endif
...@@ -229,3 +229,4 @@ cppa/detail/empty_tuple.hpp ...@@ -229,3 +229,4 @@ cppa/detail/empty_tuple.hpp
src/empty_tuple.cpp src/empty_tuple.cpp
cppa/detail/addressed_message.hpp cppa/detail/addressed_message.hpp
src/addressed_message.cpp src/addressed_message.cpp
unit_testing/test__yield_interface.cpp
...@@ -11,6 +11,8 @@ namespace cppa { namespace detail { ...@@ -11,6 +11,8 @@ namespace cppa { namespace detail {
class task_scheduler : public scheduler class task_scheduler : public scheduler
{ {
typedef scheduler super;
typedef util::single_reader_queue<scheduled_actor> job_queue; typedef util::single_reader_queue<scheduled_actor> job_queue;
job_queue m_queue; job_queue m_queue;
...@@ -21,8 +23,9 @@ class task_scheduler : public scheduler ...@@ -21,8 +23,9 @@ class task_scheduler : public scheduler
public: public:
task_scheduler(); virtual void start();
~task_scheduler();
virtual void stop();
void schedule(scheduled_actor* what); void schedule(scheduled_actor* what);
......
...@@ -25,18 +25,19 @@ struct stack_node ...@@ -25,18 +25,19 @@ struct stack_node
stack_node* next; stack_node* next;
void* s; void* s;
stack_node(): next(NULL) , s(mmap(0, STACK_SIZE, stack_node(): next(NULL) , s(/*mmap(0, STACK_SIZE,
PROT_EXEC | PROT_READ | PROT_WRITE, PROT_EXEC | PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON, MAP_PRIVATE | MAP_ANON,
-1, -1,
0)) 0)*/ new char[STACK_SIZE])
{ {
//memset(s, 0, STACK_SIZE); //memset(s, 0, STACK_SIZE);
} }
~stack_node() ~stack_node()
{ {
munmap(s, STACK_SIZE); //munmap(s, STACK_SIZE);
delete[] reinterpret_cast<char*>(s);
} }
}; };
......
...@@ -44,17 +44,17 @@ void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy) ...@@ -44,17 +44,17 @@ void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy)
} }
} }
task_scheduler::task_scheduler() void task_scheduler::start()
: m_queue()
, m_dummy()
{ {
super::start();
m_worker = thread(worker_loop, &m_queue, &m_dummy); m_worker = thread(worker_loop, &m_queue, &m_dummy);
} }
task_scheduler::~task_scheduler() void task_scheduler::stop()
{ {
m_queue.push_back(&m_dummy); m_queue.push_back(&m_dummy);
m_worker.join(); m_worker.join();
super::stop();
} }
void task_scheduler::schedule(scheduled_actor* what) void task_scheduler::schedule(scheduled_actor* what)
......
...@@ -6,6 +6,9 @@ ...@@ -6,6 +6,9 @@
#include "cppa/detail/mock_scheduler.hpp" #include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp" #include "cppa/detail/thread_pool_scheduler.hpp"
using std::cout;
using std::endl;
namespace cppa { namespace detail { namespace cppa { namespace detail {
namespace { namespace {
...@@ -28,7 +31,7 @@ struct thread_pool_scheduler::worker ...@@ -28,7 +31,7 @@ struct thread_pool_scheduler::worker
worker* next; worker* next;
bool m_done; bool m_done;
job_queue* m_job_queue; job_queue* m_job_queue;
scheduled_actor* m_job; volatile scheduled_actor* m_job;
worker_queue* m_supervisor_queue; worker_queue* m_supervisor_queue;
mutex m_mtx; mutex m_mtx;
condition_variable m_cv; condition_variable m_cv;
...@@ -51,6 +54,7 @@ struct thread_pool_scheduler::worker ...@@ -51,6 +54,7 @@ struct thread_pool_scheduler::worker
void operator()() void operator()()
{ {
// enqueue as idle worker
m_supervisor_queue->push_back(this); m_supervisor_queue->push_back(this);
// loop // loop
util::fiber fself; util::fiber fself;
...@@ -65,10 +69,84 @@ struct thread_pool_scheduler::worker ...@@ -65,10 +69,84 @@ struct thread_pool_scheduler::worker
} }
if (m_done) return; if (m_done) return;
} }
auto job = const_cast<scheduled_actor*>(m_job);
// run actor up to 300ms // run actor up to 300ms
bool reschedule = false; bool reschedule = false;
auto tout = now(); auto tout = now();
tout += std::chrono::milliseconds(300); tout += std::chrono::milliseconds(300);
set_self(job);
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, scheduled_actor::execute(m_job,
fself, fself,
[&]() -> bool [&]() -> bool
...@@ -86,11 +164,13 @@ struct thread_pool_scheduler::worker ...@@ -86,11 +164,13 @@ struct thread_pool_scheduler::worker
CPPA_MEMORY_BARRIER(); CPPA_MEMORY_BARRIER();
dec_actor_count(); dec_actor_count();
}); });
if (reschedule) */
if (reschedule && job)
{ {
m_job_queue->push_back(m_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);
} }
...@@ -108,8 +188,8 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -108,8 +188,8 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
{ {
worker_queue wqueue; worker_queue wqueue;
std::vector<worker_ptr> workers; std::vector<worker_ptr> workers;
// init // init with at least two workers
size_t num_workers = std::max<size_t>(thread::hardware_concurrency(), 1); size_t num_workers = std::max<size_t>(thread::hardware_concurrency(), 2);
auto new_worker = [&]() auto new_worker = [&]()
{ {
worker_ptr wptr(new worker(&wqueue, jqueue)); worker_ptr wptr(new worker(&wqueue, jqueue));
...@@ -124,7 +204,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -124,7 +204,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
// loop // loop
do do
{ {
// fetch job // fetch next job
scheduled_actor* job = jqueue->pop(); scheduled_actor* job = jqueue->pop();
if (job == dummy) if (job == dummy)
{ {
...@@ -132,7 +212,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -132,7 +212,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
} }
else else
{ {
// fetch waiting worker (wait up to 500ms) // fetch next idle worker (wait up to 500ms)
worker* w = nullptr; worker* w = nullptr;
auto timeout = now(); auto timeout = now();
timeout += std::chrono::milliseconds(500); timeout += std::chrono::milliseconds(500);
...@@ -154,6 +234,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -154,6 +234,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
} }
} }
while (!done); while (!done);
cout << "supervisor_loop is done\n";
// quit // quit
for (auto& w : workers) for (auto& w : workers)
{ {
......
...@@ -18,7 +18,8 @@ unit_tests_SOURCES = hash_of.cpp \ ...@@ -18,7 +18,8 @@ unit_tests_SOURCES = hash_of.cpp \
test__spawn.cpp \ test__spawn.cpp \
test__tuple.cpp \ test__tuple.cpp \
test__type_list.cpp \ test__type_list.cpp \
test__uniform_type.cpp test__uniform_type.cpp \
test__yield_interface.cpp
unit_tests_DEPENDENCIES = ../.libs/libcppa.la unit_tests_DEPENDENCIES = ../.libs/libcppa.la
......
...@@ -115,28 +115,8 @@ inline bool found_key(Iterator& i, Container& cont, Key&& key) ...@@ -115,28 +115,8 @@ inline bool found_key(Iterator& i, Container& cont, Key&& key)
return (i = cont.find(std::forward<Key>(key))) != cont.end(); return (i = cont.find(std::forward<Key>(key))) != cont.end();
} }
void print_terminator()
{
if (std::uncaught_exception())
{
try { throw; }
catch (std::exception& e)
{
cerr << "terminate called with an active exception:\n"
<< typeid(e).name() << "; what(): " << e.what() << endl;
}
catch (...)
{
cerr << "terminate called with a non-std exception" << endl;
}
}
abort();
}
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
std::set_terminate(print_terminator);
auto args = get_kv_pairs(argc, argv); auto args = get_kv_pairs(argc, argv);
if (!args.empty()) if (!args.empty())
{ {
...@@ -185,6 +165,7 @@ int main(int argc, char** argv) ...@@ -185,6 +165,7 @@ int main(int argc, char** argv)
//cout << endl << endl; //cout << endl << endl;
std::cout << std::boolalpha; std::cout << std::boolalpha;
size_t errors = 0; size_t errors = 0;
RUN_TEST(test__yield_interface);
RUN_TEST(test__ripemd_160); RUN_TEST(test__ripemd_160);
RUN_TEST(test__primitive_variant); RUN_TEST(test__primitive_variant);
RUN_TEST(test__uniform_type); RUN_TEST(test__uniform_type);
......
...@@ -48,6 +48,7 @@ std::cerr << err_msg << std::endl; \ ...@@ -48,6 +48,7 @@ std::cerr << err_msg << std::endl; \
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) CPPA_CHECK(((lhs_loc) == (rhs_loc))) #define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) CPPA_CHECK(((lhs_loc) == (rhs_loc)))
size_t test__yield_interface();
size_t test__remote_actor(const char* app_path, bool is_client, size_t test__remote_actor(const char* app_path, bool is_client,
const std::map<std::string, std::string>& args); const std::map<std::string, std::string>& args);
size_t test__ripemd_160(); size_t test__ripemd_160();
......
...@@ -22,12 +22,6 @@ void testee1() ...@@ -22,12 +22,6 @@ void testee1()
{ {
receive_loop receive_loop
( (
others() >> []()
{
any_tuple msg = last_received();
//actor_ptr sender = msg.sender();
//sender->enqueue(any_tuple(self(), sender, msg.content()));
},
after(std::chrono::milliseconds(10)) >> []() after(std::chrono::milliseconds(10)) >> []()
{ {
quit(exit_reason::user_defined); quit(exit_reason::user_defined);
...@@ -71,6 +65,17 @@ void testee3(actor_ptr parent) ...@@ -71,6 +65,17 @@ void testee3(actor_ptr parent)
size_t test__spawn() size_t test__spawn()
{ {
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
spawn(testee1);
await_all_others_done();
return CPPA_TEST_RESULT;
auto report_unexpected = [&]() auto report_unexpected = [&]()
{ {
cerr << "unexpected message: " << to_string(last_received()) << endl; cerr << "unexpected message: " << to_string(last_received()) << endl;
......
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif
#include <atomic>
#include <cstring>
#include <cstdint>
#include <type_traits>
#include <ucontext.h>
#include <sys/mman.h>
#include <signal.h>
#include <cstddef>
#include <iostream>
#include "test.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/detail/yield_interface.hpp"
using namespace cppa;
using namespace cppa::util;
using namespace cppa::detail;
namespace { ucontext_t ctx[2]; }
struct pseudo_worker
{
int m_count;
pseudo_worker() : m_count(0) { }
void operator()()
{
for (;;)
{
++m_count;
swapcontext(&ctx[1], &ctx[0]);
}
}
};
__thread pseudo_worker* t_worker = nullptr;
void coroutine()
{
(*t_worker)();
}
size_t test__yield_interface()
{
CPPA_TEST(test__yield_interface);
void* coroutine_stack = mmap(0,
SIGSTKSZ,
PROT_EXEC | PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON,
-1,
0);
pseudo_worker worker;
t_worker = &worker;
memset(&ctx[0], 0, sizeof(ucontext_t));
getcontext(&ctx[0]);
memset(&ctx[1], 0, sizeof(ucontext_t));
getcontext(&ctx[1]);
ctx[1].uc_stack.ss_sp = coroutine_stack;
ctx[1].uc_stack.ss_size = SIGSTKSZ;
ctx[1].uc_link = &ctx[0];
makecontext(&ctx[1], coroutine, 0);
auto do_switch = []() { swapcontext(&ctx[0], &ctx[1]); };
//fiber fself;
//fiber fcoroutine(coroutine, nullptr);
/*
auto do_switch = [&]() { call(&fcoroutine, &fself); };
do_switch();
CPPA_CHECK(yielded_state() == yield_state::invalid);
do_switch();
CPPA_CHECK(yielded_state() == yield_state::ready);
do_switch();
CPPA_CHECK(yielded_state() == yield_state::blocked);
do_switch();
CPPA_CHECK(yielded_state() == yield_state::done);
do_switch();
CPPA_CHECK(yielded_state() == yield_state::killed);
*/
//for (int i = 1 ; i < 11; ++i)
while (worker.m_count < 11)
{
do_switch();
}
CPPA_CHECK_EQUAL(worker.m_count, 10);
munmap(coroutine_stack, SIGSTKSZ);
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