talezero/integration
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added class Schedule for threads

Browse Source
This commit is contained in:
Josh Yelon
2021-02-18 17:21:25 -05:00
parent d931191901
commit 36ff4eacae
5 changed files with 113 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
luprex/core/cpp/world.cpp
View File

@@ -261,73 +261,60 @@ void World::invoke_plan(int64_t actor_id, int64_t place_id, const std::string &a
// Push the thread's ID into the runnable thread queue,
// then run the thread queue.
enqueue_thread(tid, place_id);
run_thread_queue();
thread_sched_.add(0, tid, place_id);
run_scheduled_threads(0);
}
void World::enqueue_thread(int64_t tid, int64_t place_id) {
thread_queue_.insert(std::make_pair(tid, place_id));
}
void World::run_thread(int64_t tid, int64_t place_id) {
void World::run_scheduled_threads(int64_t clk) {
lua_State *L = state();
LuaVar tangibles, place, mt, threads, thread;
LuaStack LS(L, tangibles, place, mt, threads, thread);
LS.rawget(tangibles, LuaRegistry, "tangibles");
LS.rawget(place, tangibles, place_id);
if (!LS.istable(place)) {
LS.result();
return;
}
LS.getmetatable(mt, place);
if (!LS.istable(mt)) {
LS.result();
return;
}
LS.rawget(threads, mt, "threads");
if (!LS.istable(threads)) {
LS.result();
return;
}
LS.rawget(thread, threads, tid);
if (!LS.isthread(thread)) {
LS.result();
return;
}
// Resume the coroutine.
lua_State *CO = LS.ckthread(thread);
int status = lua_resume(CO, 3);
while (thread_sched_.ready(clk)) {
SchedEntry sched = thread_sched_.pop();
LS.rawget(place, tangibles, sched.place_id());
if (!LS.istable(place)) {
continue;
}
LS.getmetatable(mt, place);
if (!LS.istable(mt)) {
continue;
}
LS.rawget(threads, mt, "threads");
if (!LS.istable(threads)) {
continue;
}
LS.rawget(thread, threads, sched.thread_id());
if (!LS.isthread(thread)) {
continue;
}
// Three possible outcomes: finished, yielded, or errored.
if (status == LUA_YIELD) {
// When the wait statement yields, it yields the desired timestamp.
std::cerr << "Thread yield top = " << lua_gettop(CO) << std::endl;
LS.rawset(threads, tid, LuaNil);
} else if (status == 0) {
// Successfully ran to completion. Remove from thread table.
std::cerr << "Thread ran to completion." << std::endl;
LS.rawset(threads, tid, LuaNil);
} else {
// Transfer the error message from CO to L, and add a traceback.
LS.rawset(threads, tid, LuaNil);
traceback_coroutine(L, CO);
std::cerr << lua_tostring(L, -1);
// Resume the coroutine.
lua_State *CO = LS.ckthread(thread);
int status = lua_resume(CO, 3);
// Three possible outcomes: finished, yielded, or errored.
if (status == LUA_YIELD) {
// When the wait statement yields, it yields the desired timestamp.
std::cerr << "Thread yield top = " << lua_gettop(CO) << std::endl;
// TODO: Insert the thread back into thread_sched_.
LS.rawset(threads, sched.thread_id(), LuaNil);
} else if (status == 0) {
// Successfully ran to completion. Remove from thread table.
std::cerr << "Thread ran to completion." << std::endl;
LS.rawset(threads, sched.thread_id(), LuaNil);
} else {
// Transfer the error message from CO to L, and add a traceback.
LS.rawset(threads, sched.thread_id(), LuaNil);
traceback_coroutine(L, CO);
std::cerr << lua_tostring(L, -1);
}
}
LS.result();
}
void World::run_thread_queue() {
while (!thread_queue_.empty()) {
auto iter = thread_queue_.begin();
int64_t tid = iter->first;
int64_t place_id = iter->second;
run_thread(tid, place_id);
thread_queue_.erase(iter);
}
}
LuaDefine(tangible_get, "c") {
LuaArg id;
LuaRet database;