john/cell
1
0
Fork 0
Code Issues 31 Pull Requests Packages Projects Releases 2 Wiki Activity

fix scheduler

Browse Source
This commit is contained in:
John Alanbrook
2025-12-11 14:42:30 -06:00
parent 5a3e260821
commit bf421743a5
17 changed files with 217 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
source/scheduler.c
View File

@@ -139,55 +139,59 @@ int heap_pop(timer_node *out) {
}
void *timer_thread_func(void *arg) {
while (1) {
pthread_mutex_lock(&engine.lock);
if (engine.shutting_down) {
pthread_mutex_unlock(&engine.lock);
return NULL;
}
if (arrlen(timer_heap) == 0) {
pthread_cond_wait(&engine.timer_cond, &engine.lock);
} else {
uint64_t now = cell_ns();
if (timer_heap[0].execute_at_ns <= now) {
// --- TIMER FIRED ---
timer_node t;
heap_pop(&t);
pthread_mutex_unlock(&engine.lock);
if (t.type == TIMER_NATIVE_REMOVE) {
// Execute native remove callback
actor_remove_cb(t.actor, t.timer_id, 0);
} else {
// Inject event into Actor
pthread_mutex_lock(t.actor->msg_mutex);
int idx = hmgeti(t.actor->timers, t.timer_id);
if (idx != -1) {
JSValue cb = t.actor->timers[idx].value;
hmdel(t.actor->timers, t.timer_id);
actor_clock(t.actor, cb);
JS_FreeValue(t.actor->context, cb);
}
pthread_mutex_unlock(t.actor->msg_mutex);
}
// Loop immediately to check for other expired timers
continue;
} else {
// --- WAIT FOR DEADLINE ---
struct timespec ts;
uint64_t ns = timer_heap[0].execute_at_ns;
ts.tv_sec = ns / 1000000000ULL;
ts.tv_nsec = ns % 1000000000ULL;
pthread_cond_timedwait(&engine.timer_cond, &engine.lock, &ts);
}
}
pthread_mutex_unlock(&engine.lock);
while (1) {
pthread_mutex_lock(&engine.lock);
if (engine.shutting_down) {
pthread_mutex_unlock(&engine.lock);
return NULL;
}
return NULL;
if (arrlen(timer_heap) == 0) {
pthread_cond_wait(&engine.timer_cond, &engine.lock);
} else {
uint64_t now = cell_ns();
if (timer_heap[0].execute_at_ns <= now) {
// --- TIMER FIRED ---
timer_node t;
heap_pop(&t);
pthread_mutex_unlock(&engine.lock);
if (t.type == TIMER_NATIVE_REMOVE) {
actor_remove_cb(t.actor, t.timer_id, 0);
} else {
pthread_mutex_lock(t.actor->msg_mutex);
int idx = hmgeti(t.actor->timers, t.timer_id);
if (idx != -1) {
JSValue cb = t.actor->timers[idx].value;
hmdel(t.actor->timers, t.timer_id);
actor_clock(t.actor, cb);
JS_FreeValue(t.actor->context, cb);
}
pthread_mutex_unlock(t.actor->msg_mutex);
}
continue;
} else {
// --- WAIT FOR DEADLINE ---
struct timespec ts;
uint64_t wait_ns = timer_heap[0].execute_at_ns - now;
// Convert relative wait time to absolute CLOCK_REALTIME
struct timespec now_real;
clock_gettime(CLOCK_REALTIME, &now_real);
uint64_t deadline_real_ns = (uint64_t)now_real.tv_sec * 1000000000ULL +
(uint64_t)now_real.tv_nsec + wait_ns;
ts.tv_sec = deadline_real_ns / 1000000000ULL;
ts.tv_nsec = deadline_real_ns % 1000000000ULL;
pthread_cond_timedwait(&engine.timer_cond, &engine.lock, &ts);
}
}
pthread_mutex_unlock(&engine.lock);
}
return NULL;
}
void *actor_runner(void *arg) {